Accept integers for discrete scales

CHANGELOG.md

@@ -23,6 +23,16 @@
 
 ### Fixed
 
+- `SCALE DISCRETE` now accepts integer columns, matching the documented ability
+  to force numeric data into a discrete scale. Integer columns mapped to
+  grouping aesthetics (e.g. `fill` in `DRAW bar`) can now be declared discrete
+  instead of being rejected with a validation error (#239).
+- Mapped columns dropped by a stat transform (e.g. a non-discrete `fill` on
+  `DRAW bar`) now produce a validation error naming the mapping, explaining
+  the drop, and suggesting `SCALE DISCRETE` — in both the Vega-Lite writer and
+  the renderer-backed writers, which previously failed with a generic
+  "column does not exist" message or an internal error (#239).
+
 - Free facet dimensions now resolve their domain, breaks, labels, and minor
   breaks per panel in core (`Scale::panels`, indexed by the canonical panel
   order shared by both writers). The Vega-Lite writer no longer pins the

src/execute/scale.rs

@@ -2222,4 +2222,117 @@ mod tests {
             );
         }
     }
+
+    // =========================================================================
+    // Discrete scale over integer columns (#239)
+    // =========================================================================
+
+    mod discrete_integer_tests {
+        #[cfg(feature = "duckdb")]
+        use crate::reader::DuckDBReader;
+        #[cfg(feature = "duckdb")]
+        use crate::reader::Reader;
+
+        /// An integer column mapped to fill survives the bar stat transform
+        /// when declared discrete.
+        #[cfg(feature = "duckdb")]
+        #[test]
+        fn discrete_scale_accepts_integer_fill_in_bar() {
+            let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();
+
+            let query = r#"
+                SELECT 'a' AS sex, 1 AS survived UNION ALL
+                SELECT 'a' AS sex, 0 AS survived UNION ALL
+                SELECT 'b' AS sex, 1 AS survived
+                VISUALISE sex AS x, survived AS fill
+                DRAW bar
+                SCALE DISCRETE fill
+            "#;
+
+            let spec = reader
+                .execute(query)
+                .expect("integer fill with SCALE DISCRETE should execute");
+            let layer_key = spec.plot.layers[0]
+                .data_key
+                .as_ref()
+                .expect("layer should have a data key");
+            let df = spec.data.get(layer_key).expect("layer data should exist");
+            assert!(
+                df.column("__ggsql_aes_fill__").is_ok(),
+                "fill column must survive the stat transform, got: {:?}",
+                df.schema()
+            );
+            // One row per observed x × fill combination
+            assert_eq!(df.height(), 3);
+            let fill = df.column("__ggsql_aes_fill__").unwrap();
+            let fill = arrow::compute::cast(fill, &arrow::datatypes::DataType::Utf8).unwrap();
+            let fill = fill
+                .as_any()
+                .downcast_ref::<arrow::array::StringArray>()
+                .unwrap();
+            let distinct: std::collections::HashSet<&str> = fill.iter().flatten().collect();
+            assert_eq!(distinct.len(), 2);
+        }
+
+        /// Without a discrete declaration, an integer fill dropped by the bar
+        /// stat fails with an actionable message instead of the writer's
+        /// generic "column does not exist" error.
+        #[cfg(feature = "duckdb")]
+        #[test]
+        fn dropped_integer_fill_reports_stat_drop_and_hint() {
+            let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();
+
+            let query = r#"
+                SELECT 'a' AS sex, 1 AS survived
+                VISUALISE sex AS x, survived AS fill
+                DRAW bar
+            "#;
+
+            let spec = reader
+                .execute(query)
+                .expect("execution succeeds; the drop is reported at write time");
+            use crate::writer::Writer;
+            let writer = crate::writer::vegalite::VegaLiteWriter::new();
+            let msg = match writer.write(&spec.plot, &spec.data) {
+                Err(crate::GgsqlError::ValidationError(s)) => s,
+                Err(other) => panic!("expected ValidationError, got: {}", other),
+                Ok(_) => panic!("expected error, got success"),
+            };
+            assert!(
+                msg.contains("aesthetic 'fill' mapped to column 'survived' was dropped by the stat transformation"),
+                "expected stat-drop message, got: {}",
+                msg
+            );
+            assert!(
+                msg.contains("SCALE DISCRETE fill"),
+                "expected discrete-scale hint, got: {}",
+                msg
+            );
+        }
+
+        /// Float columns are still rejected with a helpful message.
+        #[cfg(feature = "duckdb")]
+        #[test]
+        fn discrete_scale_rejects_float_fill() {
+            let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();
+
+            let query = r#"
+                SELECT 'a' AS sex, 1.5 AS score
+                VISUALISE sex AS x, score AS fill
+                DRAW bar
+                SCALE DISCRETE fill
+            "#;
+
+            let msg = match reader.execute(query) {
+                Err(crate::GgsqlError::ValidationError(s)) => s,
+                Err(other) => panic!("expected ValidationError, got: {}", other),
+                Ok(_) => panic!("expected error, got success"),
+            };
+            assert!(
+                msg.contains("floating-point"),
+                "expected floating-point message, got: {}",
+                msg
+            );
+        }
+    }
 }

src/plot/scale/scale_type/discrete.rs

@@ -25,18 +25,21 @@ impl ScaleTypeTrait for Discrete {
         match dtype {
             // Accept discrete types
             DataType::Utf8 | DataType::Boolean | DataType::Dictionary(_, _) => Ok(()),
-            // Reject numeric types
+            // Accept integer types (integers commonly encode categories: flags, codes, years)
             DataType::Int8
             | DataType::Int16
             | DataType::Int32
             | DataType::Int64
             | DataType::UInt8
             | DataType::UInt16
             | DataType::UInt32
-            | DataType::UInt64
-            | DataType::Float32
-            | DataType::Float64 => Err("Discrete scale cannot be used with numeric data. \
-                 Use CONTINUOUS or BINNED scale type instead, or ensure the column contains categorical data.".to_string()),
+            | DataType::UInt64 => Ok(()),
+            // Reject float types (use CONTINUOUS or BINNED instead)
+            DataType::Float32 | DataType::Float64 => Err(
+                "Discrete scale cannot be used with floating-point data. \
+                 Use CONTINUOUS or BINNED scale type instead."
+                    .to_string(),
+            ),
             // Reject temporal types
             DataType::Date32 => Err("Discrete scale cannot be used with Date data. \
                  Use CONTINUOUS scale type instead (dates are treated as continuous temporal data).".to_string()),
@@ -596,17 +599,26 @@ mod tests {
     }
 
     #[test]
-    fn test_validate_dtype_rejects_numeric() {
+    fn test_validate_dtype_accepts_integer() {
         use super::ScaleTypeTrait;
 
         let discrete = Discrete;
-        let result = discrete.validate_dtype(&DataType::Int64);
+        assert!(discrete.validate_dtype(&DataType::Int64).is_ok());
+        assert!(discrete.validate_dtype(&DataType::UInt8).is_ok());
+    }
+
+    #[test]
+    fn test_validate_dtype_rejects_float() {
+        use super::ScaleTypeTrait;
+
+        let discrete = Discrete;
+        let result = discrete.validate_dtype(&DataType::Float64);
         assert!(result.is_err());
         let err = result.unwrap_err();
-        assert!(err.contains("numeric"));
+        assert!(err.contains("floating-point"));
         assert!(err.contains("CONTINUOUS") || err.contains("BINNED"));
 
-        let result = discrete.validate_dtype(&DataType::Float64);
+        let result = discrete.validate_dtype(&DataType::Float32);
         assert!(result.is_err());
     }
 

src/writer/hephaestus/compose.rs

@@ -62,6 +62,12 @@ pub fn validate_plot(spec: &Plot) -> Result<()> {
 /// As [`validate_plot`] and [`build_composition`].
 pub fn prepare(spec: &Plot, data: &HashMap<String, DataFrame>) -> Result<PlotComposition> {
     validate_plot(spec)?;
+    let ctx = spec.get_aesthetic_context();
+    for (idx, layer) in spec.layers.iter().enumerate() {
+        if let Some(df) = layer.data_key.as_ref().and_then(|key| data.get(key)) {
+            crate::writer::validate_layer_columns(layer, df, idx, &ctx)?;
+        }
+    }
     build_composition(spec, data)
 }
 

src/writer/mod.rs

@@ -185,3 +185,92 @@ pub trait Writer {
         self.write(spec.plot(), spec.data())
     }
 }
+
+/// Validate column references for a single layer against its DataFrame.
+///
+/// Shared by all writers so every format reports a missing column the same
+/// way. When the layer's geom applies a stat transform, a missing mapped
+/// column almost always means the stat dropped it — only grouping columns
+/// survive a stat — so the message says that and points at `SCALE DISCRETE`
+/// rather than listing available columns.
+pub(crate) fn validate_layer_columns(
+    layer: &crate::Layer,
+    data: &DataFrame,
+    layer_idx: usize,
+    ctx: &crate::plot::aesthetic::AestheticContext,
+) -> Result<()> {
+    use crate::plot::AestheticValue;
+    use crate::{naming, GgsqlError};
+
+    let available_columns: Vec<String> = data
+        .get_column_names()
+        .iter()
+        .map(|s| s.to_string())
+        .collect();
+
+    let source_desc = || {
+        if let Some(src) = &layer.source {
+            format!(" (source: {})", src.as_str())
+        } else {
+            " (global data)".to_string()
+        }
+    };
+
+    for (aesthetic, value) in &layer.mappings.aesthetics {
+        if let AestheticValue::Column {
+            name: col,
+            original_name,
+            ..
+        } = value
+        {
+            if !available_columns.contains(col) {
+                // Translate both the column name (which may be wrapped, e.g.
+                // `__ggsql_aes_pos1__`) and the aesthetic key (which is in
+                // internal form, e.g. `pos1`) to the user-facing form.
+                let stripped = naming::extract_aesthetic_name(col).unwrap_or(col.as_str());
+                let display_col = original_name
+                    .clone()
+                    .unwrap_or_else(|| ctx.map_internal_to_user(stripped).to_string());
+                let display_aes = ctx.map_internal_to_user(aesthetic);
+                if !layer.geom.stat_consumed_aesthetics().is_empty() {
+                    return Err(GgsqlError::ValidationError(format!(
+                        "Layer {} ({}): aesthetic '{}' mapped to column '{}' was dropped by the \
+                         stat transformation. Only grouping columns survive the stat, and '{}' \
+                         is not discrete.\n\
+                         To group by it, declare the aesthetic discrete: SCALE DISCRETE {}.\n\
+                         Alternatively, convert the column to text in the SQL query.",
+                        layer_idx + 1,
+                        layer.geom,
+                        display_aes,
+                        display_col,
+                        display_col,
+                        display_aes,
+                    )));
+                }
+                return Err(GgsqlError::ValidationError(format!(
+                    "Column '{}' referenced in aesthetic '{}' (layer {}{}) does not exist.\nAvailable columns: {}",
+                    display_col,
+                    display_aes,
+                    layer_idx + 1,
+                    source_desc(),
+                    crate::and_list(&available_columns)
+                )));
+            }
+        }
+    }
+
+    // Check partition_by columns
+    for col in &layer.partition_by {
+        if !available_columns.contains(col) {
+            return Err(GgsqlError::ValidationError(format!(
+                "Column '{}' referenced in PARTITION BY (layer {}{}) does not exist.\nAvailable columns: {}",
+                col,
+                layer_idx + 1,
+                source_desc(),
+                crate::and_list(&available_columns)
+            )));
+        }
+    }
+
+    Ok(())
+}

src/writer/vegalite/layer.rs

@@ -72,57 +72,7 @@ pub fn validate_layer_columns(
     layer_idx: usize,
     ctx: &AestheticContext,
 ) -> Result<()> {
-    let available_columns: Vec<String> = data
-        .get_column_names()
-        .iter()
-        .map(|s| s.to_string())
-        .collect();
-
-    for (aesthetic, value) in &layer.mappings.aesthetics {
-        if let AestheticValue::Column { name: col, .. } = value {
-            if !available_columns.contains(col) {
-                let source_desc = if let Some(src) = &layer.source {
-                    format!(" (source: {})", src.as_str())
-                } else {
-                    " (global data)".to_string()
-                };
-                // Translate both the column name (which may be wrapped, e.g.
-                // `__ggsql_aes_pos1__`) and the aesthetic key (which is in
-                // internal form, e.g. `pos1`) to the user-facing form.
-                let stripped = naming::extract_aesthetic_name(col).unwrap_or(col.as_str());
-                let display_col = ctx.map_internal_to_user(stripped);
-                let display_aes = ctx.map_internal_to_user(aesthetic);
-                return Err(GgsqlError::ValidationError(format!(
-                    "Column '{}' referenced in aesthetic '{}' (layer {}{}) does not exist.\nAvailable columns: {}",
-                    display_col,
-                    display_aes,
-                    layer_idx + 1,
-                    source_desc,
-                    crate::and_list(&available_columns)
-                )));
-            }
-        }
-    }
-
-    // Check partition_by columns
-    for col in &layer.partition_by {
-        if !available_columns.contains(col) {
-            let source_desc = if let Some(src) = &layer.source {
-                format!(" (source: {})", src.as_str())
-            } else {
-                " (global data)".to_string()
-            };
-            return Err(GgsqlError::ValidationError(format!(
-                "Column '{}' referenced in PARTITION BY (layer {}{}) does not exist.\nAvailable columns: {}",
-                col,
-                layer_idx + 1,
-                source_desc,
-                crate::and_list(&available_columns)
-            )));
-        }
-    }
-
-    Ok(())
+    crate::writer::validate_layer_columns(layer, data, layer_idx, ctx)
 }
 
 // =============================================================================