ggsql 0.4.0

//! Geom rendering for Vega-Lite writer
//!
//! This module provides:
//! - Basic geom-to-mark mapping and column validation
//! - A trait-based approach to rendering different ggsql geom types to Vega-Lite specs
//!
//! Each geom type can override specific phases of the rendering pipeline while using
//! sensible defaults for standard behavior.

use crate::plot::aesthetic::AestheticContext;
use crate::plot::layer::geom::GeomType;
use crate::plot::layer::is_transposed;
use crate::plot::{ArrayElement, ParameterValue};
use crate::writer::vegalite::POINTS_TO_PIXELS;
use crate::{naming, AestheticValue, DataFrame, Geom, GgsqlError, Layer, Result};
use arrow::array::Array;
use serde_json::{json, Map, Value};
use std::any::Any;
use std::collections::HashMap;

use super::data::{dataframe_to_values, dataframe_to_values_with_bins, ROW_INDEX_COLUMN};
use super::encoding::RenderContext;

// =============================================================================
// Basic Geom Utilities
// =============================================================================

/// Map ggsql Geom to Vega-Lite mark type
/// Always includes `clip: true` to ensure marks don't render outside plot bounds
pub fn geom_to_mark(geom: &Geom) -> Value {
    let mark_type = match geom.geom_type() {
        GeomType::Point => "point",
        GeomType::Line => "line",
        GeomType::Path => "line",
        GeomType::Bar => "bar",
        GeomType::Area => "area",
        GeomType::Tile => "rect",
        GeomType::Ribbon => "area",
        GeomType::Polygon => "line",
        GeomType::Histogram => "bar",
        GeomType::Density => "area",
        GeomType::Violin => "line",
        GeomType::Boxplot => "boxplot",
        GeomType::Text => "text",
        GeomType::Segment => "rule",
        GeomType::Smooth => "line",
        GeomType::Rule => "rule",
        GeomType::Range => "rule",
        GeomType::Spatial => "geoshape",
        _ => "point", // Default fallback
    };
    json!({
        "type": mark_type,
        "clip": true
    })
}

/// Map a `side` value to a positive/negative sign in the orientation-aware way
/// shared by violin, boxplot, and (effectively) jitter rendering. Returns true
/// if `side` falls on the positive offset half (right/top of a vertical layer,
/// bottom/left of a horizontal layer). Caller is responsible for handling
/// `"both"` separately.
fn side_is_positive(side: &str, is_horizontal: bool) -> bool {
    if is_horizontal {
        matches!(side, "bottom" | "left")
    } else {
        matches!(side, "top" | "right")
    }
}

/// Validate column references for a single layer against its specific DataFrame
pub fn validate_layer_columns(
    layer: &Layer,
    data: &DataFrame,
    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(())
}

// =============================================================================
// GeomRenderer Trait System
// =============================================================================

/// Data prepared for a layer - either single dataset or multiple components
pub enum PreparedData {
    /// Standard single dataset (most geoms)
    Single {
        values: Vec<Value>,
        metadata: Box<dyn Any + Send + Sync>,
    },
    /// Multiple component datasets (boxplot, violin, range)
    Composite {
        components: HashMap<String, Vec<Value>>,
        metadata: Box<dyn Any + Send + Sync>,
    },
}

// =============================================================================
// GeomRenderer Trait System
// =============================================================================

/// Trait for rendering ggsql geoms to Vega-Lite layers
///
/// Provides a three-phase rendering pipeline:
/// 1. **Data Preparation**: Convert DataFrame to JSON values
/// 2. **Encoding Modifications**: Apply geom-specific encoding transformations
/// 3. **Layer Output**: Finalize and potentially expand layers
///
/// Most geoms use the default implementations. Only geoms with special requirements
/// (bar width, path ordering, boxplot decomposition) need to override specific methods.
pub trait GeomRenderer {
    // === Phase 1: Data Preparation ===

    /// Prepare data for this layer.
    /// Default: convert DataFrame to JSON values (single dataset)
    fn prepare_data(
        &self,
        df: &DataFrame,
        _layer: &Layer,
        _data_key: &str,
        binned_columns: &HashMap<String, Vec<f64>>,
    ) -> Result<PreparedData> {
        let values = if binned_columns.is_empty() {
            dataframe_to_values(df)?
        } else {
            dataframe_to_values_with_bins(df, binned_columns)?
        };
        Ok(PreparedData::Single {
            values,
            metadata: Box::new(()),
        })
    }

    // === Phase 2: Encoding Modifications ===

    /// Modify the encoding map for this geom.
    /// Default: no modifications
    fn modify_encoding(
        &self,
        _encoding: &mut Map<String, Value>,
        _layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        Ok(())
    }

    /// Modify the mark/layer spec for this geom.
    /// Default: no modifications
    fn modify_spec(
        &self,
        _layer_spec: &mut Value,
        _layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        Ok(())
    }

    // === Phase 3: Layer Output ===

    /// Whether to add the standard source filter transform.
    /// Default: true (composite geoms override to return false)
    fn needs_source_filter(&self) -> bool {
        true
    }

    /// Finalize the layer(s) for output.
    /// Default: return single layer unchanged
    /// Composite geoms override to expand into multiple layers
    fn finalize(
        &self,
        layer_spec: Value,
        _layer: &Layer,
        _data_key: &str,
        _prepared: &PreparedData,
        _context: &RenderContext,
    ) -> Result<Vec<Value>> {
        Ok(vec![layer_spec])
    }
}

// =============================================================================
// Default Renderer (for geoms with no special handling)
// =============================================================================

/// Default renderer used for geoms with standard behavior
pub struct DefaultRenderer;

impl GeomRenderer for DefaultRenderer {}

// =============================================================================
// Point Renderer
// =============================================================================

/// Renderer for the `point` geom.
///
/// Works around a Vega-Lite quirk: when `mark.filled` is omitted/`true` and the
/// top-level `opacity` channel is unset, the renderer applies its own implicit
/// opacity multiplier on top of `fillOpacity`/`strokeOpacity` — so even with
/// `fillOpacity: 1.0`, the points appear semitransparent.
///
/// The fix:
/// - Force `mark.filled = false`. Custom SVG shape paths (which ggsql always
///   emits) are still rendered as filled regardless of this setting, so this
///   only suppresses the implicit opacity behavior.
/// - Force `encoding.opacity = {value: 1.0}`. With the global opacity pinned
///   to 1.0, `fillOpacity` and `strokeOpacity` (set from the user's `opacity =>`
///   setting upstream) become the actual opacity controls.
pub struct PointRenderer;

impl GeomRenderer for PointRenderer {
    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        _layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        if let Some(mark) = layer_spec.get_mut("mark") {
            if let Some(obj) = mark.as_object_mut() {
                obj.insert("filled".to_string(), json!(false));
            }
        }
        if let Some(encoding) = layer_spec.get_mut("encoding") {
            if let Some(obj) = encoding.as_object_mut() {
                obj.insert("opacity".to_string(), json!({"value": 1.0}));
            }
        }
        Ok(())
    }
}

// =============================================================================
// Bar Renderer
// =============================================================================

/// Renderer for bar geom - overrides mark spec for band width
pub struct BarRenderer;

impl GeomRenderer for BarRenderer {
    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        layer: &Layer,
        context: &RenderContext,
    ) -> Result<()> {
        let width = match layer.adjusted_width {
            // The adjusted width comes from position adjustments
            Some(adjusted) => adjusted,
            _ => match layer.parameters.get("width") {
                // Fallback to width parameter value if there is no adjustment
                Some(ParameterValue::Number(n)) => *n,
                _ => 0.9,
            },
        };

        // For horizontal bars, use "height" for band size; for vertical, use "width"
        let is_horizontal = is_transposed(layer);
        let (pos1, _, _, pos2, _, _) = &context.channels;
        let axis = if is_horizontal { pos2 } else { pos1 };

        let size_value = match layer_spec["encoding"][axis]["bin"].as_str() {
            // I don't think binned scales obey 'band', but they don't tolerate the 'expr' option.
            Some("binned") => json!({"band": width}),
            // Use expression-based size with the adjusted width
            _ => json!({"expr": format!("bandwidth('{}') * {}", axis, width)}),
        };

        layer_spec["mark"] = if is_horizontal {
            json!({
                "type": "bar",
                "height": size_value,
                "baseline": "middle",
                "clip": true
            })
        } else {
            json!({
                "type": "bar",
                "width": size_value,
                "align": "center",
                "clip": true
            })
        };
        Ok(())
    }
}

// =============================================================================
// Path Renderer
// =============================================================================

/// Renderer for path and line geoms - preserves data order for correct rendering
///
/// Automatically detects when continuous material aesthetics (stroke, linewidth, opacity) vary
/// within partition groups and converts to segmented rendering using detail encoding.
/// Discrete material aesthetics (linetype, or discrete stroke) already define groups
/// via partition_by and don't require special handling.
///
/// Handles both `line` and `path` geoms - the only difference is the mark type used.
pub struct PathRenderer;

// =============================================================================
// Helper functions for path/line segmentation
// =============================================================================

/// Find row indices where any of the specified columns change value.
///
/// Returns a sorted vector starting with 0 (first row), followed by indices
/// where any column value differs from the previous row. Does not include
/// the final boundary (n_rows).
///
/// Used by both line segmentation and text font run-length encoding.
fn find_change_starts(df: &DataFrame, columns: &[String]) -> Result<Vec<usize>> {
    use crate::array_util::value_to_string;

    let n_rows = df.height();

    if columns.is_empty() || n_rows <= 1 {
        return Ok(vec![0]);
    }

    // Build a change mask manually: for each row i (1..n), check if any column differs from row i-1
    let mut change_starts = vec![0];

    for i in 1..n_rows {
        let mut changed = false;
        for col_name in columns {
            let col = df.column(col_name).map_err(|e| {
                GgsqlError::InternalError(format!("Column '{}' not found: {}", col_name, e))
            })?;

            // Compare values using string representation
            let curr_null = col.is_null(i);
            let prev_null = col.is_null(i - 1);

            if curr_null != prev_null {
                changed = true;
                break;
            }
            if !curr_null {
                let curr_val = value_to_string(col, i);
                let prev_val = value_to_string(col, i - 1);
                if curr_val != prev_val {
                    changed = true;
                    break;
                }
            }
        }
        if changed {
            change_starts.push(i);
        }
    }

    Ok(change_starts)
}

/// Check if an aesthetic varies within any group segment
///
/// Uses precomputed group boundaries to efficiently check if the aesthetic
/// has multiple distinct values within any group segment.
fn aesthetic_varies_within_groups(
    df: &DataFrame,
    aesthetic_col: &str,
    group_boundaries: &[usize],
) -> Result<bool> {
    use crate::array_util::value_to_string;
    use std::collections::HashSet;

    let col = df.column(aesthetic_col).map_err(|e| {
        GgsqlError::InternalError(format!("Column '{}' not found: {}", aesthetic_col, e))
    })?;

    // Check each group segment
    for window in group_boundaries.windows(2) {
        let start = window[0];
        let end = window[1];

        if end - start < 2 {
            continue; // Single-row groups can't vary
        }

        // Count unique values in this segment
        let mut unique = HashSet::new();
        for i in start..end {
            if col.is_null(i) {
                unique.insert("__null__".to_string());
            } else {
                unique.insert(value_to_string(col, i));
            }
            if unique.len() > 1 {
                return Ok(true);
            }
        }
    }

    Ok(false)
}

impl GeomRenderer for PathRenderer {
    fn prepare_data(
        &self,
        df: &DataFrame,
        layer: &Layer,
        _data_key: &str,
        binned_columns: &HashMap<String, Vec<f64>>,
    ) -> Result<PreparedData> {
        // Continuous material aesthetics that can trigger segmentation
        // (linetype is always discrete and already handled via partition_by)
        let material_aesthetics: &[&'static str] = &["stroke", "linewidth", "opacity"];

        // Start with existing partition_by (includes discrete material aesthetics already)
        let partition_columns: Vec<String> = layer.partition_by.clone();

        // Compute group boundaries based on existing partitions
        let n_rows = df.height();
        let group_boundaries = if partition_columns.is_empty() || n_rows <= 1 {
            vec![0, n_rows]
        } else {
            let mut boundaries = find_change_starts(df, &partition_columns)?;
            boundaries.push(n_rows);
            boundaries
        };

        // Check continuous material aesthetics (not in partition_by) for within-group variation
        let mut varying_aesthetics: Vec<&'static str> = Vec::new();

        for &aesthetic in material_aesthetics {
            if let Some(AestheticValue::Column { name: col, .. }) = layer.mappings.get(aesthetic) {
                // Skip if already in partition_by (discrete, already defines groups)
                if !layer.partition_by.contains(col) {
                    // Continuous: check if varies within groups
                    if aesthetic_varies_within_groups(df, col, &group_boundaries)? {
                        varying_aesthetics.push(aesthetic);
                        // Don't add to partition_columns - continuous values shouldn't partition
                    }
                }
            }
        }

        // Return the data with segmentation metadata if needed
        let values = if binned_columns.is_empty() {
            dataframe_to_values(df)?
        } else {
            dataframe_to_values_with_bins(df, binned_columns)?
        };

        Ok(PreparedData::Single {
            values,
            metadata: Box::new(varying_aesthetics),
        })
    }

    fn modify_encoding(
        &self,
        encoding: &mut Map<String, Value>,
        _layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        // Use row index field to preserve natural data order
        // (we've already ordered in SQL via apply_stat_transform)
        encoding.insert(
            "order".to_string(),
            json!({"field": ROW_INDEX_COLUMN, "type": "quantitative"}),
        );
        Ok(())
    }

    fn finalize(
        &self,
        mut layer_spec: Value,
        layer: &Layer,
        _data_key: &str,
        prepared: &PreparedData,
        context: &RenderContext,
    ) -> Result<Vec<Value>> {
        // Get metadata from prepared data
        let PreparedData::Single { metadata, .. } = prepared else {
            return Err(GgsqlError::InternalError(
                "PathRenderer expects PreparedData::Single".to_string(),
            ));
        };

        // Get varying aesthetics from metadata
        let Some(varying_aesthetics) = metadata.downcast_ref::<Vec<&'static str>>() else {
            return Ok(vec![layer_spec]);
        };

        // Handle varying linewidth: switch to trail mark and translate encodings
        if varying_aesthetics.contains(&"linewidth") {
            layer_spec["mark"] = json!({"type": "trail", "clip": true, "strokeWidth": 0});

            // Translate line encodings to trail encodings
            if let Some(encoding_obj) = layer_spec.get_mut("encoding") {
                if let Some(encoding_map) = encoding_obj.as_object_mut() {
                    // strokeWidth → size
                    if let Some(stroke_width) = encoding_map.remove("strokeWidth") {
                        encoding_map.insert("size".to_string(), stroke_width);
                    }

                    // stroke → fill
                    if let Some(mut stroke) = encoding_map.remove("stroke") {
                        // Add symbolStrokeColor to legend so symbols display with color
                        if let Some(stroke_obj) = stroke.as_object_mut() {
                            if let Some(legend) = stroke_obj.get_mut("legend") {
                                if let Some(legend_obj) = legend.as_object_mut() {
                                    legend_obj.insert(
                                        "symbolStrokeColor".to_string(),
                                        json!({"expr": "scale('fill', datum.value)"}),
                                    );
                                }
                            }
                        }
                        encoding_map.insert("fill".to_string(), stroke);
                    }

                    // opacity → fillOpacity
                    if let Some(opacity) = encoding_map.remove("opacity") {
                        encoding_map.insert("fillOpacity".to_string(), opacity);
                    }
                }
            }
        }

        // Handle varying stroke/opacity: apply segmentation
        if !varying_aesthetics.contains(&"stroke") && !varying_aesthetics.contains(&"opacity") {
            // Only linewidth varies, trail mark handles it natively
            return Ok(vec![layer_spec]);
        }

        // Build list of fields to segment (always pos1/pos2, plus size if linewidth varies)
        let (pos1, _, _, pos2, _, _) = &context.channels;
        let mut segment_fields = vec![
            (pos1.as_str(), naming::aesthetic_column("pos1")),
            (pos2.as_str(), naming::aesthetic_column("pos2")),
        ];
        if varying_aesthetics.contains(&"linewidth") {
            segment_fields.push(("size", naming::aesthetic_column("linewidth")));
        }

        // Segmented rendering using detail encoding:
        // 1. Create segment IDs (row_index serves as segment ID)
        // 2. Create next row's values using window transform
        // 3. Flatten to create 2 rows per segment (point_index: 0=start, 1=end)
        // 4. Use calculate to pick current or next based on point_index
        // 5. Add segment ID to detail encoding

        // Preserve existing transforms (e.g., source filter)
        let mut transforms = layer_spec
            .get("transform")
            .and_then(|t| t.as_array())
            .cloned()
            .unwrap_or_default();

        // Step 1 & 2: Window transform to get next row's values
        let window_ops: Vec<Value> = segment_fields
            .iter()
            .map(|(_, field)| {
                json!({
                    "op": "lead",
                    "field": field,
                    "as": format!("{}_next", field)
                })
            })
            .collect();

        let mut window_transform = json!({
            "window": window_ops,
            "sort": [{"field": ROW_INDEX_COLUMN}]
        });

        if !layer.partition_by.is_empty() {
            window_transform["groupby"] = json!(layer.partition_by);
        }

        transforms.push(window_transform);

        // Step 2b: Filter out last row in each group (no next point)
        // Check the first field (x) for null to detect end of segments
        let first_field = &segment_fields[0].1;
        transforms.push(json!({
            "filter": format!("datum.{}_next != null", first_field)
        }));

        // Step 3: Flatten to create 2 rows per segment
        // Create a constant array [0, 1] to flatten
        transforms.push(json!({
            "calculate": "[0, 1]",
            "as": "__segment_points__"
        }));

        transforms.push(json!({
            "flatten": ["__segment_points__"],
            "as": ["__point_index__"]
        }));

        // Step 4: Calculate actual field values based on point_index
        for (_, field) in &segment_fields {
            transforms.push(json!({
                "calculate": format!("datum.__point_index__ == 0 ? datum.{} : datum.{}_next", field, field),
                "as": format!("{}_final", field)
            }));
        }

        // Step 5: Create segment ID (use original row_index)
        transforms.push(json!({
            "calculate": format!("datum.{}", ROW_INDEX_COLUMN),
            "as": "__segment_id__"
        }));

        layer_spec["transform"] = json!(transforms);
        // Don't set layer_spec["data"] - use the unified top-level dataset
        // The source filter transform will select the correct rows

        // Update encodings to use final field values and add segment_id to detail
        if let Some(encoding_obj) = layer_spec.get_mut("encoding") {
            if let Some(encoding_map) = encoding_obj.as_object_mut() {
                // Update each field encoding to use _final
                for (encoding_name, field) in &segment_fields {
                    if let Some(enc) = encoding_map.get_mut(*encoding_name) {
                        if let Some(enc_obj) = enc.as_object_mut() {
                            enc_obj.insert("field".to_string(), json!(format!("{}_final", field)));
                        }
                    }
                }

                // Add segment_id to detail encoding
                encoding_map.insert(
                    "detail".to_string(),
                    json!({
                        "field": "__segment_id__",
                        "type": "nominal"
                    }),
                );
            }
        }

        Ok(vec![layer_spec])
    }
}

// =============================================================================
// Rule Renderer
// =============================================================================

pub struct RuleRenderer;

impl GeomRenderer for RuleRenderer {
    fn modify_encoding(
        &self,
        encoding: &mut Map<String, Value>,
        layer: &Layer,
        context: &RenderContext,
    ) -> Result<()> {
        // Check if this is a diagonal rule (slope is non-zero)
        let diagonal = matches!(
            layer.parameters.get("diagonal"),
            Some(ParameterValue::Boolean(true))
        );

        if !diagonal {
            // Regular horizontal/vertical rule - no special rendering needed
            return Ok(());
        }
        // Use layer's pre-computed orientation
        let (pos1, pos1_end, _, pos2, pos2_end, _) = &context.channels;
        let (primary, primary2, secondary, secondary2, extent_aes) = if is_transposed(layer) {
            (pos2, pos2_end, pos1, pos1_end, "pos2")
        } else {
            (pos1, pos1_end, pos2, pos2_end, "pos1")
        };

        // Get primary axis extent from context to set explicit scale domain
        // This prevents an axis drift
        let mut primary_enco = json!({"field": "primary_min", "type": "quantitative"});
        if let Ok((min, max)) = context.get_extent(extent_aes) {
            primary_enco["scale"] = json!({"domain": [min, max]})
        };

        // Add encodings for rule mark
        // primary_min/primary_max are created by transforms (extent of the axis)
        // secondary_min/secondary_max are computed via formula
        encoding.insert(primary.clone(), primary_enco);
        encoding.insert(
            primary2.clone(),
            json!({
                "field": "primary_max"
            }),
        );
        encoding.insert(
            secondary.clone(),
            json!({
                "field": "secondary_min",
                "type": "quantitative"
            }),
        );
        encoding.insert(
            secondary2.clone(),
            json!({
                "field": "secondary_max"
            }),
        );

        Ok(())
    }

    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        layer: &Layer,
        context: &RenderContext,
    ) -> Result<()> {
        // Densified rule: expanded to a multi-row line under map projection
        if matches!(
            layer.parameters.get("densified"),
            Some(ParameterValue::Boolean(true))
        ) {
            layer_spec["mark"] = json!({
                "type": "line",
                "clip": true
            });

            if let Some(encoding) = layer_spec
                .get_mut("encoding")
                .and_then(|e| e.as_object_mut())
            {
                encoding.insert(
                    "order".to_string(),
                    json!({"field": ROW_INDEX_COLUMN, "type": "quantitative"}),
                );
            }

            return Ok(());
        }

        // Determine slope expression: either a literal value or a field reference
        let slope_expr = match layer.mappings.get("slope") {
            Some(AestheticValue::Literal(ParameterValue::Number(n))) if *n == 0.0 => {
                // Slope is 0 - no diagonal
                None
            }
            Some(AestheticValue::Literal(ParameterValue::Number(n))) => {
                // Literal non-zero slope - inline the value
                Some(n.to_string())
            }
            Some(AestheticValue::Column { .. }) | Some(AestheticValue::AnnotationColumn { .. }) => {
                // Column-based slope - reference the field
                let slope_field = naming::aesthetic_column("slope");
                Some(format!("datum.{}", slope_field))
            }
            _ => {
                // No slope mapping - no diagonal
                None
            }
        };

        let Some(slope_expr) = slope_expr else {
            return Ok(());
        };

        let (intercept_field, extent_aes) = if is_transposed(layer) {
            // x is intercept
            (naming::aesthetic_column("pos1"), "pos2")
        } else {
            // y is intercept
            (naming::aesthetic_column("pos2"), "pos1")
        };

        // Get extent from appropriate axis:
        let (primary_min, primary_max) = context.get_extent(extent_aes)?;

        // Add transforms:
        // 1. Create constant primary_min/primary_max fields (extent of the primary axis)
        // 2. Compute secondary values at those primary positions: secondary = slope * primary + intercept
        //    (where intercept is pos1 for x-mapped or pos2 for y-mapped)
        let transforms = json!([
            {
                "calculate": primary_min.to_string(),
                "as": "primary_min"
            },
            {
                "calculate": primary_max.to_string(),
                "as": "primary_max"
            },
            {
                "calculate": format!("{} * datum.primary_min + datum.{}", slope_expr, intercept_field),
                "as": "secondary_min"
            },
            {
                "calculate": format!("{} * datum.primary_max + datum.{}", slope_expr, intercept_field),
                "as": "secondary_max"
            }
        ]);

        // Prepend to existing transforms (if any)
        if let Some(existing) = layer_spec.get("transform") {
            if let Some(arr) = existing.as_array() {
                let mut new_transforms = transforms.as_array().unwrap().clone();
                new_transforms.extend_from_slice(arr);
                layer_spec["transform"] = json!(new_transforms);
            }
        } else {
            layer_spec["transform"] = transforms;
        }

        Ok(())
    }
}

// =============================================================================
// Text Renderer
// =============================================================================

/// Renderer for text geom - handles font properties via data splitting
pub struct TextRenderer;

impl TextRenderer {
    /// Analyze DataFrame columns to build font property runs using run-length encoding.
    /// Returns:
    /// - DataFrame where each row represents a run's font properties (family, fontweight, italic, hjust, vjust, angle)
    /// - Vec<usize> of run lengths corresponding to each row
    fn build_font_rle(df: &DataFrame) -> Result<(DataFrame, Vec<usize>)> {
        use arrow::array::ArrayRef;
        use arrow::compute;

        let nrows = df.height();

        if nrows == 0 {
            // Return empty DataFrame and empty run lengths
            return Ok((DataFrame::empty(), Vec::new()));
        }

        // Collect font property column names that exist in the DataFrame
        let font_aesthetics = [
            "typeface",
            "fontweight",
            "italic",
            "hjust",
            "vjust",
            "rotation",
        ];

        let mut font_column_names = Vec::new();
        let mut font_columns: HashMap<&str, &ArrayRef> = HashMap::new();

        for aesthetic in font_aesthetics {
            let col_name = naming::aesthetic_column(aesthetic);
            if let Ok(col) = df.column(&col_name) {
                font_column_names.push(col_name);
                font_columns.insert(aesthetic, col);
            }
        }

        // Find indices where any font property changes
        let change_indices = find_change_starts(df, &font_column_names)?;

        // Calculate run lengths
        let run_lengths: Vec<usize> = change_indices
            .iter()
            .enumerate()
            .map(|(i, &start)| {
                let end = change_indices.get(i + 1).copied().unwrap_or(nrows);
                end - start
            })
            .collect();

        // Extract rows at change indices (only font columns) using arrow take
        let indices_array: ArrayRef = std::sync::Arc::new(arrow::array::UInt32Array::from(
            change_indices
                .iter()
                .map(|&i| i as u32)
                .collect::<Vec<u32>>(),
        ));

        let mut result_cols: Vec<(String, ArrayRef)> = Vec::new();
        for aesthetic in font_aesthetics {
            if let Some(col) = font_columns.get(aesthetic) {
                let taken = compute::take(
                    col.as_ref(),
                    indices_array
                        .as_any()
                        .downcast_ref::<arrow::array::UInt32Array>()
                        .unwrap(),
                    None,
                )
                .map_err(|e| {
                    GgsqlError::InternalError(format!(
                        "Failed to take indices from {}: {}",
                        aesthetic, e
                    ))
                })?;
                result_cols.push((naming::aesthetic_column(aesthetic), taken));
            }
        }

        // Create result DataFrame (only font properties, no run_length column)
        let result_df = DataFrame::new(result_cols)?;

        Ok((result_df, run_lengths))
    }

    /// Split label values containing newlines into arrays of strings
    ///
    /// Uses the shared split_label_on_newlines function to ensure consistent
    /// newline handling across all label types (text data, axis labels, titles, etc.)
    fn split_label_newlines(values: &mut [Value]) -> Result<()> {
        let label_col = naming::aesthetic_column("label");

        for row in values.iter_mut() {
            // Get the object, skip if not an object
            let Some(obj) = row.as_object_mut() else {
                continue;
            };

            // Get the label value, skip if not present
            let Some(label_value) = obj.get(&label_col) else {
                continue;
            };
            // Get the string value, skip if not a string
            let Some(label_str) = label_value.as_str() else {
                continue;
            };

            // Use shared function for consistent newline splitting
            obj.insert(label_col.clone(), super::split_label_on_newlines(label_str));
        }
        Ok(())
    }

    /// Convert typeface to Vega-Lite font value
    /// Prefers literal over column value
    fn convert_typeface(
        literal: Option<&ParameterValue>,
        column_value: Option<&str>,
    ) -> Option<Value> {
        // First select which value to use (prefer literal)
        let value = if let Some(ParameterValue::String(s)) = literal {
            s.as_str()
        } else {
            column_value?
        };

        // Then apply conversion
        if !value.is_empty() {
            Some(json!(value))
        } else {
            None
        }
    }

    /// Convert fontweight to Vega-Lite fontWeight value
    /// Prefers literal over column value
    /// Accepts all CSS font-weight keywords and numeric values:
    /// - Keywords: 'thin', 'hairline', 'extra-light', 'ultra-light', 'light',
    ///   'normal', 'regular', 'lighter', 'medium', 'semi-bold', 'demi-bold',
    ///   'bold', 'bolder', 'extra-bold', 'ultra-bold', 'black', 'heavy'
    /// - Numeric values: any number
    /// - Numeric strings from columns: '100', '400', '700', etc.
    ///
    /// Always outputs 'normal' or 'bold' for Vega-Lite compatibility:
    /// - < 500 → 'normal' (thin, light, normal, regular, lighter)
    /// - >= 500 → 'bold' (medium, semi-bold, bold, bolder, extra-bold, black, heavy)
    fn convert_fontweight(
        literal: Option<&ParameterValue>,
        column_value: Option<&str>,
    ) -> Option<Value> {
        // First select which value to use (prefer literal)
        let numeric = match literal {
            Some(ParameterValue::String(s)) => {
                // String literal: keyword or numeric string
                Self::parse_fontweight_to_numeric(s.as_str())
            }
            Some(ParameterValue::Number(n)) => {
                // Numeric literal: use directly
                Some(*n)
            }
            _ => {
                // Column value: try to parse
                column_value.and_then(Self::parse_fontweight_to_numeric)
            }
        }?;

        // Apply >= 500 rule to determine bold/normal
        let is_bold = numeric >= 500.0;
        Some(json!(if is_bold { "bold" } else { "normal" }))
    }

    /// Parse fontweight value from string to numeric value
    fn parse_fontweight_to_numeric(value: &str) -> Option<f64> {
        // Try parsing as number first
        if let Ok(num) = value.parse::<f64>() {
            return Some(num);
        }

        // Map CSS font-weight keywords to numeric values
        // Normalize: convert to lowercase and remove hyphens for flexible matching
        let normalized = value.to_lowercase().replace("-", "");
        match normalized.as_str() {
            "thin" | "hairline" => Some(100.0),
            "extralight" | "ultralight" => Some(200.0),
            "light" => Some(300.0),
            "normal" | "regular" | "lighter" => Some(400.0),
            "medium" => Some(500.0),
            "semibold" | "demibold" => Some(600.0),
            "bold" | "bolder" => Some(700.0),
            "extrabold" | "ultrabold" => Some(800.0),
            "black" | "heavy" => Some(900.0),
            _ => None,
        }
    }

    /// Convert italic to Vega-Lite fontStyle value
    /// Prefers literal over column value
    /// Accepts boolean literals or string column values ('true', 'false', '1', '0')
    fn convert_italic(
        literal: Option<&ParameterValue>,
        column_value: Option<&str>,
    ) -> Option<Value> {
        // First select which value to use (prefer literal)
        let value = if let Some(ParameterValue::Boolean(b)) = literal {
            *b
        } else if let Some(s) = column_value {
            // Parse string to boolean
            match s.to_lowercase().as_str() {
                "true" | "1" => true,
                "false" | "0" => false,
                _ => return None,
            }
        } else {
            return None;
        };

        // Convert boolean to fontStyle
        let style = if value { "italic" } else { "normal" };
        Some(json!(style))
    }

    /// Convert hjust to Vega-Lite align value
    /// Prefers literal over column value
    fn convert_hjust(
        literal: Option<&ParameterValue>,
        column_value: Option<&str>,
    ) -> Option<Value> {
        // First extract which value to use (prefer literal)
        let value_str = match literal {
            Some(ParameterValue::String(s)) => s.to_string(),
            Some(ParameterValue::Number(n)) => n.to_string(),
            _ => column_value?.to_string(),
        };

        // Then apply conversion inline
        let align = match value_str.parse::<f64>() {
            Ok(v) if v <= 0.25 => "left",
            Ok(v) if v >= 0.75 => "right",
            _ => match value_str.as_str() {
                "left" => "left",
                "right" => "right",
                _ => "center",
            },
        };

        Some(json!(align))
    }

    /// Convert vjust to Vega-Lite baseline value
    /// Prefers literal over column value
    fn convert_vjust(
        literal: Option<&ParameterValue>,
        column_value: Option<&str>,
    ) -> Option<Value> {
        // First extract which value to use (prefer literal)
        let value_str = match literal {
            Some(ParameterValue::String(s)) => s.to_string(),
            Some(ParameterValue::Number(n)) => n.to_string(),
            _ => column_value?.to_string(),
        };

        // Then apply conversion inline
        let baseline = match value_str.parse::<f64>() {
            Ok(v) if v <= 0.25 => "bottom",
            Ok(v) if v >= 0.75 => "top",
            _ => match value_str.as_str() {
                "top" => "top",
                "bottom" => "bottom",
                _ => "middle",
            },
        };

        Some(json!(baseline))
    }

    /// Convert rotation to Vega-Lite angle value (degrees)
    /// Prefers literal over column value
    /// Normalizes angles to [0, 360) range
    fn convert_rotation(
        literal: Option<&ParameterValue>,
        column_value: Option<f64>,
    ) -> Option<Value> {
        // First select which value to use (prefer literal)
        let value = if let Some(ParameterValue::Number(n)) = literal {
            *n
        } else {
            column_value?
        };

        // Then apply conversion inline
        let normalized = value % 360.0;
        let angle = if normalized < 0.0 {
            normalized + 360.0
        } else {
            normalized
        };

        Some(json!(angle))
    }

    /// Apply font properties to mark object from DataFrame row and layer literals
    /// Uses literals from layer parameters if present, otherwise uses DataFrame column values
    fn apply_font_properties(
        mark_obj: &mut Map<String, Value>,
        df: &DataFrame,
        row_idx: usize,
        layer: &Layer,
    ) -> Result<()> {
        // Helper to extract string column values using aesthetic column naming
        let get_str = |aesthetic: &str| -> Option<String> {
            use crate::array_util::as_str;
            let col_name = naming::aesthetic_column(aesthetic);
            let col = df.column(&col_name).ok()?;
            if col.is_null(row_idx) {
                return None;
            }
            as_str(col).ok().map(|ca| ca.value(row_idx).to_string())
        };

        // Helper to extract numeric column values (for angle)
        let get_f64 = |aesthetic: &str| -> Option<f64> {
            use crate::array_util::{as_f64, as_str, cast_array};
            use arrow::datatypes::DataType;
            let col_name = naming::aesthetic_column(aesthetic);
            let col = df.column(&col_name).ok()?;

            if col.is_null(row_idx) {
                return None;
            }

            // Try as string first (for string-encoded numbers)
            if let Ok(ca) = as_str(col) {
                return ca.value(row_idx).parse::<f64>().ok();
            }

            // Try as numeric types directly
            if let Ok(casted) = cast_array(col, &DataType::Float64) {
                if let Ok(ca) = as_f64(&casted) {
                    return Some(ca.value(row_idx));
                }
            }

            None
        };

        // Convert and apply font properties
        if let Some(typeface_val) = Self::convert_typeface(
            layer.get_literal("typeface"),
            get_str("typeface").as_deref(),
        ) {
            mark_obj.insert("font".to_string(), typeface_val);
        }

        if let Some(weight) = Self::convert_fontweight(
            layer.get_literal("fontweight"),
            get_str("fontweight").as_deref(),
        ) {
            mark_obj.insert("fontWeight".to_string(), weight);
        }

        if let Some(style) =
            Self::convert_italic(layer.get_literal("italic"), get_str("italic").as_deref())
        {
            mark_obj.insert("fontStyle".to_string(), style);
        }

        if let Some(hjust_val) =
            Self::convert_hjust(layer.get_literal("hjust"), get_str("hjust").as_deref())
        {
            mark_obj.insert("align".to_string(), hjust_val);
        }

        if let Some(vjust_val) =
            Self::convert_vjust(layer.get_literal("vjust"), get_str("vjust").as_deref())
        {
            mark_obj.insert("baseline".to_string(), vjust_val);
        }

        if let Some(angle_val) =
            Self::convert_rotation(layer.get_literal("rotation"), get_f64("rotation"))
        {
            mark_obj.insert("angle".to_string(), angle_val);
        }

        Ok(())
    }

    /// Build transform with source filter
    fn build_transform_with_filter(prototype: &Value, source_key: &str) -> Vec<Value> {
        let source_filter = json!({
            "filter": {
                "field": naming::SOURCE_COLUMN,
                "equal": source_key
            }
        });

        let existing_transforms = prototype
            .get("transform")
            .and_then(|t| t.as_array())
            .cloned()
            .unwrap_or_default();

        let mut new_transforms = vec![source_filter];
        new_transforms.extend(existing_transforms);
        new_transforms
    }

    /// Finalize layers as nested layer with shared encoding (works for single or multiple runs)
    fn finalize_nested_layers(
        &self,
        prototype: Value,
        data_key: &str,
        font_runs_df: &DataFrame,
        run_lengths: &[usize],
        layer: &Layer,
    ) -> Result<Vec<Value>> {
        // Extract shared encoding from prototype
        let shared_encoding = prototype.get("encoding").cloned();

        // Build base mark object with fixed parameters
        let mut base_mark = json!({"type": "text"});
        if let Some(mark_map) = base_mark.as_object_mut() {
            // Extract offset parameter (offset => [x, y] or offset => n)
            match layer.parameters.get("offset") {
                Some(ParameterValue::Array(offset_array)) if offset_array.len() == 2 => {
                    // Array case: [x, y]
                    if let ArrayElement::Number(x_offset) = offset_array[0] {
                        mark_map.insert("xOffset".to_string(), json!(x_offset * POINTS_TO_PIXELS));
                    }
                    if let ArrayElement::Number(y_offset) = offset_array[1] {
                        mark_map.insert("yOffset".to_string(), json!(-y_offset * POINTS_TO_PIXELS));
                    }
                }
                Some(ParameterValue::Number(offset)) => {
                    // Single number case: applies to both x and y
                    mark_map.insert("xOffset".to_string(), json!(offset * POINTS_TO_PIXELS));
                    mark_map.insert("yOffset".to_string(), json!(-offset * POINTS_TO_PIXELS));
                }
                _ => {}
            }
        }

        // Build individual layers without encoding (mark + transform only)
        // Use run_lengths to get number of runs (works even when no font columns exist)
        let nruns = run_lengths.len();
        let mut nested_layers: Vec<Value> = Vec::with_capacity(nruns);

        for run_idx in 0..nruns {
            let suffix = format!("_font_{}", run_idx);
            let source_key = format!("{}{}", data_key, suffix);

            // Clone base mark and apply font-specific properties
            let mut mark_obj = base_mark.clone();
            if let Some(mark_map) = mark_obj.as_object_mut() {
                Self::apply_font_properties(mark_map, font_runs_df, run_idx, layer)?;
            }

            // Create layer with mark and transform (no encoding)
            nested_layers.push(json!({
                "mark": mark_obj,
                "transform": Self::build_transform_with_filter(&prototype, &source_key)
            }));
        }

        // Wrap in parent spec with shared encoding
        let mut parent_spec = json!({"layer": nested_layers});

        if let Some(encoding) = shared_encoding {
            parent_spec["encoding"] = encoding;
        }

        Ok(vec![parent_spec])
    }
}

impl GeomRenderer for TextRenderer {
    fn prepare_data(
        &self,
        df: &DataFrame,
        _layer: &Layer,
        _data_key: &str,
        binned_columns: &HashMap<String, Vec<f64>>,
    ) -> Result<PreparedData> {
        // Note: Label formatting is already applied via Text::post_process() during execution

        // Analyze font columns to get RLE runs
        let (font_runs_df, run_lengths) = Self::build_font_rle(df)?;

        // Split data by font runs, tracking cumulative position
        let mut components: HashMap<String, Vec<Value>> = HashMap::new();
        let mut position = 0;

        for (run_idx, &length) in run_lengths.iter().enumerate() {
            let suffix = format!("_font_{}", run_idx);

            // Slice the contiguous run from the DataFrame (more efficient than boolean masking)
            let sliced = df.slice(position, length);

            let mut values = if binned_columns.is_empty() {
                dataframe_to_values(&sliced)?
            } else {
                dataframe_to_values_with_bins(&sliced, binned_columns)?
            };

            // Post-process label values to split on newlines
            Self::split_label_newlines(&mut values)?;

            components.insert(suffix, values);
            position += length;
        }

        Ok(PreparedData::Composite {
            components,
            metadata: Box::new((font_runs_df, run_lengths)),
        })
    }

    fn modify_encoding(
        &self,
        encoding: &mut Map<String, Value>,
        _layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        // Remove font aesthetics from encoding - they only work as mark properties
        for &aesthetic in &[
            "typeface",
            "fontweight",
            "italic",
            "hjust",
            "vjust",
            "rotation",
        ] {
            encoding.remove(aesthetic);
        }

        // Suppress legend and scale for text encoding
        if let Some(text_encoding) = encoding.get_mut("text") {
            if let Some(text_obj) = text_encoding.as_object_mut() {
                text_obj.insert("legend".to_string(), Value::Null);
                text_obj.insert("scale".to_string(), Value::Null);
            }
        }

        Ok(())
    }

    fn needs_source_filter(&self) -> bool {
        // TextRenderer handles source filtering in finalize()
        false
    }

    fn finalize(
        &self,
        prototype: Value,
        layer: &Layer,
        data_key: &str,
        prepared: &PreparedData,
        _context: &RenderContext,
    ) -> Result<Vec<Value>> {
        let PreparedData::Composite { metadata, .. } = prepared else {
            return Err(GgsqlError::InternalError(
                "TextRenderer::finalize called with non-composite data".to_string(),
            ));
        };

        // Downcast metadata to font runs
        let (font_runs_df, run_lengths) = metadata
            .downcast_ref::<(DataFrame, Vec<usize>)>()
            .ok_or_else(|| GgsqlError::InternalError("Failed to downcast font runs".to_string()))?;

        // Generate nested layers from font runs (works for single or multiple runs)
        self.finalize_nested_layers(prototype, data_key, font_runs_df, run_lengths, layer)
    }
}

// =============================================================================
// Tile Renderer
// =============================================================================

/// Renderer for tile geom - handles continuous and discrete rectangles
///
/// For continuous scales: remaps xmin/xmax → x/x2, ymin/ymax → y/y2
/// For discrete scales: keeps x/y as-is and applies width/height as band fractions
pub struct TileRenderer;

impl GeomRenderer for TileRenderer {
    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        layer: &Layer,
        context: &RenderContext,
    ) -> Result<()> {
        let (pos1, pos1_end, _, pos2, pos2_end, _) = &context.channels;

        // Polygonized tile: densified rectangle rendered as closed line
        if matches!(
            layer.parameters.get("densified"),
            Some(ParameterValue::Boolean(true))
        ) {
            layer_spec["mark"] = json!({
                "type": "line",
                "interpolate": "linear-closed"
            });

            if let Some(encoding) = layer_spec
                .get_mut("encoding")
                .and_then(|e| e.as_object_mut())
            {
                // Preserve vertex order
                encoding.insert(
                    "order".to_string(),
                    json!({"field": ROW_INDEX_COLUMN, "type": "quantitative"}),
                );
            }

            return Ok(());
        }

        // Discrete tile: rect mark with band-based width/height
        let encoding = layer_spec
            .get_mut("encoding")
            .and_then(|e| e.as_object_mut());

        let Some(encoding) = encoding else {
            return Ok(());
        };

        // Check which directions are discrete
        let pos1_is_discrete = !encoding.contains_key(pos1_end.as_str());
        let pos2_is_discrete = !encoding.contains_key(pos2_end.as_str());

        // Early return if both continuous (standard rect mark is fine)
        if !pos1_is_discrete && !pos2_is_discrete {
            return Ok(());
        }

        // Build mark properties for discrete directions
        let mut mark = json!({
            "type": "rect",
            "clip": true
        });

        if pos1_is_discrete {
            if let Some(width_enc) = encoding.remove("width") {
                if let Some(field) = width_enc.get("field").and_then(|f| f.as_str()) {
                    // Field encoding: use expression with datum reference
                    mark["width"] = json!({
                        "expr": format!("datum.{} * bandwidth('{}')", field, pos1)
                    });
                } else if let Some(value) = width_enc.get("value") {
                    // Literal value: use band syntax
                    mark["width"] = json!({"band": value});
                }
            }
        }

        if pos2_is_discrete {
            if let Some(height_enc) = encoding.remove("height") {
                if let Some(field) = height_enc.get("field").and_then(|f| f.as_str()) {
                    // Field encoding: use expression with datum reference
                    mark["height"] = json!({
                        "expr": format!("datum.{} * bandwidth('{}')", field, pos2)
                    });
                } else if let Some(value) = height_enc.get("value") {
                    // Literal value: use band syntax
                    mark["height"] = json!({"band": value});
                }
            }
        }

        if mark.get("width").is_some() || mark.get("height").is_some() {
            layer_spec["mark"] = mark;
        }

        Ok(())
    }
}

// =============================================================================
// Polygon Renderer
// =============================================================================

/// Renderer for polygon geom - uses closed line with fill
pub struct PolygonRenderer;

impl GeomRenderer for PolygonRenderer {
    fn modify_encoding(
        &self,
        encoding: &mut Map<String, Value>,
        _layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        // Polygon needs both `fill` and `stroke` independently, but map_aesthetic_name()
        // converts fill -> color (which works for most geoms). For closed line marks,
        // we need actual `fill` and `stroke` channels, so we undo the mapping here.
        if let Some(color) = encoding.remove("color") {
            encoding.insert("fill".to_string(), color);
        }
        // Use row index field to preserve natural data order
        encoding.insert(
            "order".to_string(),
            json!({"field": ROW_INDEX_COLUMN, "type": "quantitative"}),
        );
        Ok(())
    }

    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        _layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        layer_spec["mark"] = json!({
            "type": "line",
            "interpolate": "linear-closed"
        });
        Ok(())
    }
}

// =============================================================================
// Segment Renderer
// =============================================================================

/// Renderer for segment geom — normally a rule (x/y/x2/y2), but when densified
/// under map projection the segment is expanded to a multi-row line.
pub struct SegmentRenderer;

impl GeomRenderer for SegmentRenderer {
    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        layer: &Layer,
        context: &RenderContext,
    ) -> Result<()> {
        if matches!(
            layer.parameters.get("densified"),
            Some(ParameterValue::Boolean(true))
        ) {
            layer_spec["mark"] = json!({
                "type": "line",
                "clip": true
            });

            if let Some(encoding) = layer_spec
                .get_mut("encoding")
                .and_then(|e| e.as_object_mut())
            {
                let (_, pos1_end, _, _, pos2_end, _) = &context.channels;
                encoding.remove(pos1_end.as_str());
                encoding.remove(pos2_end.as_str());
                encoding.insert(
                    "order".to_string(),
                    json!({"field": ROW_INDEX_COLUMN, "type": "quantitative"}),
                );
            }
        }

        Ok(())
    }
}

// =============================================================================
// Ribbon Renderer
// =============================================================================

/// Renderer for ribbon geom — normally an area (x/y/y2), but when densified
/// under map projection the ribbon is expanded to a closed polygon.
pub struct RibbonRenderer;

impl GeomRenderer for RibbonRenderer {
    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        layer: &Layer,
        context: &RenderContext,
    ) -> Result<()> {
        if matches!(
            layer.parameters.get("densified"),
            Some(ParameterValue::Boolean(true))
        ) {
            layer_spec["mark"] = json!({
                "type": "line",
                "interpolate": "linear-closed"
            });

            if let Some(encoding) = layer_spec
                .get_mut("encoding")
                .and_then(|e| e.as_object_mut())
            {
                let (_, pos1_end, _, _, pos2_end, _) = &context.channels;
                encoding.remove(pos1_end.as_str());
                encoding.remove(pos2_end.as_str());
                encoding.insert(
                    "order".to_string(),
                    json!({"field": ROW_INDEX_COLUMN, "type": "quantitative"}),
                );
            }
        }

        Ok(())
    }
}

// =============================================================================
// Violin Renderer
// =============================================================================

/// Renderer for violin geom - uses line
pub struct ViolinRenderer;

impl GeomRenderer for ViolinRenderer {
    fn modify_spec(
        &self,
        layer_spec: &mut Value,
        layer: &Layer,
        _context: &RenderContext,
    ) -> Result<()> {
        layer_spec["mark"] = json!({
            "type": "line",
            "filled": true
        });
        let offset_col = naming::aesthetic_column("offset");

        // Read orientation from layer (already resolved during execution)
        let is_horizontal = is_transposed(layer);

        // It'll be implemented as an offset.
        let violin_offset = match layer.parameters.get("side") {
            Some(ParameterValue::String(side)) if side != "both" => {
                if side_is_positive(side, is_horizontal) {
                    format!("[datum.{offset}]", offset = offset_col)
                } else {
                    format!("[-datum.{offset}]", offset = offset_col)
                }
            }
            _ => format!("[datum.{offset}, -datum.{offset}]", offset = offset_col),
        };

        // Continuous axis column for order calculation:
        // - Vertical: pos2 (y-axis has continuous density values)
        // - Horizontal: pos1 (x-axis has continuous density values)
        let continuous_col = if is_horizontal {
            naming::aesthetic_column("pos1")
        } else {
            naming::aesthetic_column("pos2")
        };

        // We use an order calculation to create a proper closed shape.
        // Right side (+ offset), sort by -continuous (top -> bottom)
        // Left side (- offset), sort by +continuous (bottom -> top)
        let calc_order = format!(
            "datum.__violin_offset > 0 ? -datum.{} : datum.{}",
            continuous_col, continuous_col
        );

        // Preserve existing transforms (e.g., source filter) and extend with violin-specific transforms
        let existing_transforms = layer_spec
            .get("transform")
            .and_then(|t| t.as_array())
            .cloned()
            .unwrap_or_default();

        // Combine the violin offset with the dodge offset (if any). Dodge
        // stores its per-row offset in the *categorical* axis offset column,
        // which is pos1offset in vertical orientation and pos2offset in
        // horizontal orientation (where the categorical axis has been
        // flipped to pos2 by orientation resolution).
        let dodge_offset_col = if is_horizontal {
            naming::aesthetic_column("pos2offset")
        } else {
            naming::aesthetic_column("pos1offset")
        };

        let mut transforms = existing_transforms;
        transforms.extend(vec![
            json!({
                // Mirror offset on both sides (offset is pre-scaled to [0, 0.5 * width])
                "calculate": violin_offset,
                "as": "violin_offsets"
            }),
            json!({
                "flatten": ["violin_offsets"],
                "as": ["__violin_offset"]
            }),
            json!({
                // Add the dodge displacement (if any) so the violin is
                // positioned correctly within its dodged group.
                "calculate": format!(
                    "datum.{dodge} != null ? datum.__violin_offset + datum.{dodge} : datum.__violin_offset",
                    dodge = dodge_offset_col
                ),
                "as": "__final_offset"
            }),
            json!({
                "calculate": calc_order,
                "as": "__order"
            }),
        ]);

        layer_spec["transform"] = json!(transforms);
        Ok(())
    }

    fn modify_encoding(
        &self,
        encoding: &mut Map<String, Value>,
        layer: &Layer,
        context: &RenderContext,
    ) -> Result<()> {
        // Read orientation from layer (already resolved during execution)
        let is_horizontal = is_transposed(layer);
        let (pos1, _, pos1_offset, pos2, _, pos2_offset) = &context.channels;

        encoding.remove("offset");

        // Categorical axis for detail encoding:
        // - Vertical: pos1 channel (categorical groups)
        // - Horizontal: pos2 channel (categorical groups)
        let categorical_channel = if is_horizontal { pos2 } else { pos1 };

        // Ensure categorical field is in detail encoding to create separate violins per category
        // This is needed because line marks with filled:true require detail to create separate paths
        let categorical_field = encoding
            .get(categorical_channel.as_str())
            .and_then(|x| x.get("field"))
            .and_then(|f| f.as_str())
            .map(|s| s.to_string());

        if let Some(cat_field) = categorical_field {
            match encoding.get_mut("detail") {
                Some(detail)
                    if detail.is_object()
                    // Single field object - check if it's already the categorical field, otherwise convert to array
                    && detail.get("field").and_then(|f| f.as_str()) != Some(&cat_field) =>
                {
                    let existing = detail.clone();
                    *detail = json!([existing, {"field": cat_field, "type": "nominal"}]);
                }
                Some(detail) if detail.is_array() => {
                    // Array - check if categorical field already present, add if not
                    let arr = detail.as_array_mut().unwrap();
                    let has_cat = arr
                        .iter()
                        .any(|d| d.get("field").and_then(|f| f.as_str()) == Some(&cat_field));
                    if !has_cat {
                        arr.push(json!({"field": cat_field, "type": "nominal"}));
                    }
                }
                None => {
                    // No detail encoding - add it with categorical field
                    encoding.insert(
                        "detail".to_string(),
                        json!({"field": cat_field, "type": "nominal"}),
                    );
                }
                _ => {}
            }
        }

        // Violins use filled line marks, which don't show a fill in the legend.
        // We intercept the encoding to pupulate a different symbol to display
        for aesthetic in ["fill", "stroke"] {
            if let Some(channel) = encoding.get_mut(aesthetic) {
                // Skip if legend is explicitly null or if it's a literal value
                if channel.get("legend").is_some_and(|v| v.is_null()) {
                    continue;
                }
                if channel.get("value").is_some() {
                    continue;
                }

                // Add/update legend properties
                let legend = channel.get_mut("legend").and_then(|v| v.as_object_mut());
                if let Some(legend_map) = legend {
                    legend_map.insert("symbolType".to_string(), json!("circle"));
                } else {
                    channel["legend"] = json!({
                        "symbolType": "circle"
                    });
                }
            }
        }

        // Offset channel based on orientation
        let offset_channel = if is_horizontal {
            pos2_offset
        } else {
            pos1_offset
        };
        encoding.insert(
            offset_channel.clone(),
            json!({
                "field": "__final_offset",
                "type": "quantitative",
                "scale": {
                    "domain": [-0.5, 0.5]
                }
            }),
        );
        encoding.insert(
            "order".to_string(),
            json!({
                "field": "__order",
                "type": "quantitative"
            }),
        );
        Ok(())
    }
}

// =============================================================================
// Range Renderer
// =============================================================================

struct RangeRenderer;

impl GeomRenderer for RangeRenderer {
    fn finalize(
        &self,
        layer_spec: Value,
        layer: &Layer,
        _data_key: &str,
        _prepared: &PreparedData,
        context: &RenderContext,
    ) -> Result<Vec<Value>> {
        // Get hinge parameter (in points)
        let width = if let Some(ParameterValue::Number(num)) = layer.parameters.get("hinge") {
            (*num) * POINTS_TO_PIXELS
        } else {
            // If no hinge specified, return just the main error bar without hinges
            return Ok(vec![layer_spec]);
        };

        let mut layers = vec![layer_spec.clone()];

        // Determine orientation and which axis holds the error range.
        // Transposition flips DataFrame columns upstream: pos2min↔pos1min, pos2max↔pos1max.
        let (pos1, pos1_end, _, pos2, pos2_end, _) = &context.channels;
        let is_vertical = !is_transposed(layer);
        let (orient, position, min_field, max_field) = if is_vertical {
            (
                "horizontal",
                pos2,
                naming::aesthetic_column("pos2min"),
                naming::aesthetic_column("pos2max"),
            )
        } else {
            (
                "vertical",
                pos1,
                naming::aesthetic_column("pos1min"),
                naming::aesthetic_column("pos1max"),
            )
        };

        // First hinge (at min position)
        let mut hinge = layer_spec.clone();
        hinge["mark"] = json!({
            "type": "tick",
            "orient": orient,
            "size": width,
            "thickness": 0,
            "clip": true
        });
        hinge["encoding"][position]["field"] = json!(min_field);
        // Remove end channels (not needed for tick mark)
        if let Some(e) = hinge["encoding"].as_object_mut() {
            e.remove(pos1_end.as_str());
            e.remove(pos2_end.as_str());
        }
        layers.push(hinge.clone());

        // Second hinge (at max position) - reuse first hinge and only change position field
        hinge["encoding"][position]["field"] = json!(max_field);
        layers.push(hinge);

        Ok(layers)
    }
}

// =============================================================================
// Boxplot Renderer
// =============================================================================

/// Metadata for boxplot rendering
struct BoxplotMetadata {
    /// Whether there are any outliers
    has_outliers: bool,
}

/// Renderer for boxplot geom - splits into multiple component layers
pub struct BoxplotRenderer;

impl BoxplotRenderer {
    /// Prepare boxplot data by splitting into type-specific datasets.
    ///
    /// Returns a HashMap of type_suffix -> data_values, plus has_outliers flag.
    /// Type suffixes are: "lower_whisker", "upper_whisker", "box", "median", "outlier"
    fn prepare_components(
        &self,
        data: &DataFrame,
        binned_columns: &HashMap<String, Vec<f64>>,
    ) -> Result<(HashMap<String, Vec<Value>>, bool)> {
        let type_col = naming::aesthetic_column("type");
        let type_col = type_col.as_str();

        // Get the type column for filtering
        let type_array = data
            .column(type_col)
            .map_err(|e| GgsqlError::WriterError(e.to_string()))?;
        let type_str_array = crate::array_util::as_str(type_array)
            .map_err(|e| GgsqlError::WriterError(e.to_string()))?;

        // Check for outliers
        let has_outliers = (0..type_str_array.len())
            .any(|i| !type_str_array.is_null(i) && type_str_array.value(i) == "outlier");

        // Split data by type into separate datasets
        let mut type_datasets: HashMap<String, Vec<Value>> = HashMap::new();

        for type_name in &["lower_whisker", "upper_whisker", "box", "median", "outlier"] {
            // Collect row indices matching this type
            let matching_indices: Vec<usize> = (0..type_str_array.len())
                .filter(|&i| !type_str_array.is_null(i) && type_str_array.value(i) == *type_name)
                .collect();

            // Skip empty datasets (e.g., no outliers)
            if matching_indices.is_empty() {
                continue;
            }

            // Take matching rows, then drop the type column.
            let indices = arrow::array::UInt32Array::from(
                matching_indices
                    .iter()
                    .map(|&i| i as u32)
                    .collect::<Vec<u32>>(),
            );
            let filtered = data
                .take(&indices)
                .and_then(|df| df.drop(type_col))
                .map_err(|e| {
                    GgsqlError::WriterError(format!("Failed to build filtered DataFrame: {}", e))
                })?;

            let values = if binned_columns.is_empty() {
                dataframe_to_values(&filtered)?
            } else {
                dataframe_to_values_with_bins(&filtered, binned_columns)?
            };

            type_datasets.insert(type_name.to_string(), values);
        }

        Ok((type_datasets, has_outliers))
    }

    /// Render boxplot layers using filter transforms on the unified dataset.
    ///
    /// Creates 5 layers: outliers (optional), lower whiskers, upper whiskers, box, median line.
    fn render_layers(
        &self,
        prototype: Value,
        layer: &Layer,
        base_key: &str,
        has_outliers: bool,
        context: &RenderContext,
    ) -> Result<Vec<Value>> {
        let mut layers: Vec<Value> = Vec::new();

        // Read orientation from layer (already resolved during execution)
        let is_horizontal = is_transposed(layer);

        // Value columns depend on orientation (after DataFrame column flip):
        // - Vertical: values in pos2/pos2end (no flip)
        // - Horizontal: values in pos1/pos1end (was pos2/pos2end before flip)
        let (value_col, value2_col) = if is_horizontal {
            (
                naming::aesthetic_column("pos1"),
                naming::aesthetic_column("pos1end"),
            )
        } else {
            (
                naming::aesthetic_column("pos2"),
                naming::aesthetic_column("pos2end"),
            )
        };

        // Validate pos1 aesthetic exists (required for boxplot)
        layer
            .mappings
            .get("pos1")
            .and_then(|x| x.column_name())
            .ok_or_else(|| {
                GgsqlError::WriterError("Boxplot requires 'x' aesthetic mapping".to_string())
            })?;
        // Validate pos2 aesthetic exists (required for boxplot)
        layer
            .mappings
            .get("pos2")
            .and_then(|y| y.column_name())
            .ok_or_else(|| {
                GgsqlError::WriterError("Boxplot requires 'y' aesthetic mapping".to_string())
            })?;

        let (pos1, pos1_end, pos1_offset, pos2, pos2_end, pos2_offset) = &context.channels;
        let value_var1 = if is_horizontal { pos1 } else { pos2 };
        let value_var2 = if is_horizontal { pos1_end } else { pos2_end };
        let axis = if is_horizontal { pos2 } else { pos1 };

        // Get width parameter
        let width = match layer.adjusted_width {
            // The adjusted width comes from position adjustments
            Some(adjusted) => adjusted,
            _ => match layer.parameters.get("width") {
                // Fallback to width parameter value if there is no adjustment
                Some(ParameterValue::Number(n)) => *n,
                _ => 0.9,
            },
        };
        let width_value = json!({"expr": format!("bandwidth('{}') * {}", axis, width)});

        // Helper to create filter transform for source selection
        let make_source_filter = |type_suffix: &str| -> Value {
            let source_key = format!("{}{}", base_key, type_suffix);
            json!({
                "filter": {
                    "field": naming::SOURCE_COLUMN,
                    "equal": source_key
                }
            })
        };

        // Helper to create a layer with source filter and mark
        let create_layer = |proto: &Value, type_suffix: &str, mark: Value| -> Value {
            let mut layer_spec = proto.clone();
            let existing_transforms = layer_spec
                .get("transform")
                .and_then(|t| t.as_array())
                .cloned()
                .unwrap_or_default();
            let mut new_transforms = vec![make_source_filter(type_suffix)];
            new_transforms.extend(existing_transforms);
            layer_spec["transform"] = json!(new_transforms);
            layer_spec["mark"] = mark;
            layer_spec
        };

        // Create outlier points layer (if there are outliers)
        if has_outliers {
            let mut points = create_layer(
                &prototype,
                "outlier",
                json!({
                    "type": "point"
                }),
            );
            if points["encoding"].get("color").is_some() {
                points["mark"]["filled"] = json!(true);
            }

            layers.push(points);
        }

        // Clone prototype without size/shape (these apply only to points)
        let mut summary_prototype = prototype.clone();
        if let Some(Value::Object(ref mut encoding)) = summary_prototype.get_mut("encoding") {
            encoding.remove("size");
            encoding.remove("shape");
        }

        // Build encoding templates for y and y2 fields
        let mut y_encoding = summary_prototype["encoding"][value_var1].clone();
        y_encoding["field"] = json!(value_col);
        // y2 is a secondary position channel — Vega-Lite only allows field/datum/value
        let y2_encoding = json!({"field": value2_col});

        // Lower whiskers (rule from y to y2, where y=q1 and y2=lower)
        let mut lower_whiskers = create_layer(
            &summary_prototype,
            "lower_whisker",
            json!({
                "type": "rule"
            }),
        );

        // Handle strokeWidth -> size for rule marks
        if let Some(linewidth) = lower_whiskers["encoding"].get("strokeWidth").cloned() {
            lower_whiskers["encoding"]["size"] = linewidth;
            if let Some(Value::Object(ref mut encoding)) = lower_whiskers.get_mut("encoding") {
                encoding.remove("strokeWidth");
            }
        }

        lower_whiskers["encoding"][value_var1] = y_encoding.clone();
        lower_whiskers["encoding"][value_var2] = y2_encoding.clone();

        // Upper whiskers (rule from y to y2, where y=q3 and y2=upper)
        let mut upper_whiskers = create_layer(
            &summary_prototype,
            "upper_whisker",
            json!({
                "type": "rule"
            }),
        );

        // Handle strokeWidth -> size for rule marks
        if let Some(linewidth) = upper_whiskers["encoding"].get("strokeWidth").cloned() {
            upper_whiskers["encoding"]["size"] = linewidth;
            if let Some(Value::Object(ref mut encoding)) = upper_whiskers.get_mut("encoding") {
                encoding.remove("strokeWidth");
            }
        }

        upper_whiskers["encoding"][value_var1] = y_encoding.clone();
        upper_whiskers["encoding"][value_var2] = y2_encoding.clone();

        // Resolve the `side` parameter. When `side != "both"`, the box and
        // median tick render at half their normal size and anchor to the
        // categorical centerline so they only occupy one side of each band.
        // Whiskers and outliers stay on the centerline. The dodge-aware
        // `width` value is unchanged so a half-box still occupies half of
        // its dodge group's allocated space — this is what lets a half-box
        // pair with a half-violin or one-sided jitter on the same band.
        let side = layer
            .parameters
            .get("side")
            .and_then(|v| match v {
                ParameterValue::String(s) => Some(s.as_str()),
                _ => None,
            })
            .unwrap_or("both");
        let half_side = side != "both";
        let side_positive = half_side && side_is_positive(side, is_horizontal);

        // For `side != "both"`, halve the bar width and shift the bar to
        // one side of the band. We reuse the same mechanism dodge already
        // uses to shift bars within a band: an `xOffset` (or `yOffset`)
        // encoding pointing at a numeric column whose value is interpreted
        // as a band-fraction shift via a scale with domain `[-0.5, 0.5]`.
        //
        // The box/median needs to be shifted by `±w/4` of the bandwidth so
        // a half-width bar (`bandwidth*w/2`) sits centred on the middle of
        // its chosen half-band. When dodge is also active there is already
        // a `pos1offset` (or `pos2offset`) column carrying the dodge shift
        // — we add the side shift to that and point the box/median at the
        // combined column. Whiskers and outliers continue to use the
        // original pos1offset (or none), so they stay centred on the
        // dodge centerline as required.
        let box_size = if half_side {
            json!({"expr": format!("bandwidth('{}') * {} / 2", axis, width)})
        } else {
            width_value.clone()
        };
        let side_shift: f64 = if side_positive {
            width / 4.0
        } else {
            -width / 4.0
        };
        let size_key = if is_horizontal { "height" } else { "width" };

        // Pick the categorical-axis offset channel and column based on
        // orientation. The categorical axis is pos1 (vertical) or pos2
        // (horizontal); the offset channel is pos1_offset / pos2_offset.
        let (offset_channel, base_offset_col) = if is_horizontal {
            (pos2_offset.as_str(), naming::aesthetic_column("pos2offset"))
        } else {
            (pos1_offset.as_str(), naming::aesthetic_column("pos1offset"))
        };
        let combined_offset_col = "__ggsql_box_side_offset__".to_string();

        // Helper that, when half_side is true, adds a calculate transform
        // to combine the dodge offset (if any) with the side shift, then
        // points the layer's offset encoding at the combined column.
        let apply_side_shift = |layer_spec: &mut Value| {
            if !half_side {
                return;
            }
            let calc_expr = format!(
                "(datum[\"{base}\"] != null ? datum[\"{base}\"] : 0) + {shift}",
                base = base_offset_col,
                shift = side_shift,
            );
            let existing = layer_spec
                .get("transform")
                .and_then(|t| t.as_array())
                .cloned()
                .unwrap_or_default();
            let mut transforms = existing;
            transforms.push(json!({
                "calculate": calc_expr,
                "as": combined_offset_col,
            }));
            layer_spec["transform"] = json!(transforms);

            layer_spec["encoding"][offset_channel] = json!({
                "field": combined_offset_col,
                "type": "quantitative",
                "scale": {"domain": [-0.5, 0.5]},
            });
        };

        // Box (bar from y to y2, where y=q1 and y2=q3)
        let mut box_part = create_layer(
            &summary_prototype,
            "box",
            json!({
                "type": "bar",
                size_key: box_size,
            }),
        );
        box_part["encoding"][value_var1] = y_encoding.clone();
        box_part["encoding"][value_var2] = y2_encoding.clone();
        apply_side_shift(&mut box_part);

        // Median line (tick at y, where y=median) — same width and shift as the box
        let mut median_line = create_layer(
            &summary_prototype,
            "median",
            json!({
                "type": "tick",
                size_key: box_size,
            }),
        );
        median_line["encoding"][value_var1] = y_encoding.clone();
        apply_side_shift(&mut median_line);

        layers.push(lower_whiskers);
        layers.push(upper_whiskers);

        // Whisker cap hinges: tick marks at the whisker tips
        if let Some(ParameterValue::Number(hinge_pts)) = layer.parameters.get("hinge") {
            let hinge_size = if half_side {
                hinge_pts * POINTS_TO_PIXELS / 2.0
            } else {
                hinge_pts * POINTS_TO_PIXELS
            };
            let orient = if is_horizontal {
                "vertical"
            } else {
                "horizontal"
            };

            let hinge_mark = json!({
                "type": "tick",
                "orient": orient,
                "size": hinge_size,
                "thickness": 0,
                "clip": true
            });

            let mut enco = y_encoding.clone();
            enco["field"] = json!(value2_col);

            let mut lower_hinge =
                create_layer(&summary_prototype, "lower_whisker", hinge_mark.clone());
            lower_hinge["encoding"][value_var1] = enco.clone();
            if let Some(Value::Object(ref mut enc)) = lower_hinge.get_mut("encoding") {
                enc.remove(value_var2);
            }

            let mut upper_hinge = create_layer(&summary_prototype, "upper_whisker", hinge_mark);
            upper_hinge["encoding"][value_var1] = enco;
            if let Some(Value::Object(ref mut enc)) = upper_hinge.get_mut("encoding") {
                enc.remove(value_var2);
            }

            // When half_side, shift the hinge tick toward the chosen side
            // using the offset channel. Convert the pixel shift to a band
            // fraction via bandwidth() so it uses the same scale as dodge.
            if half_side {
                let sign = if side_positive { 1.0 } else { -1.0 };
                let hinge_offset_col = "__ggsql_hinge_offset__";
                let calc_expr = format!(
                    "(datum[\"{base}\"] != null ? datum[\"{base}\"] : 0) + {sign} * {px} / bandwidth('{axis}')",
                    base = base_offset_col,
                    sign = sign,
                    px = hinge_size / 2.0,
                    axis = axis,
                );
                let apply_hinge_offset = |layer_spec: &mut Value| {
                    let existing = layer_spec
                        .get("transform")
                        .and_then(|t| t.as_array())
                        .cloned()
                        .unwrap_or_default();
                    let mut transforms = existing;
                    transforms.push(json!({
                        "calculate": calc_expr,
                        "as": hinge_offset_col,
                    }));
                    layer_spec["transform"] = json!(transforms);
                    layer_spec["encoding"][offset_channel] = json!({
                        "field": hinge_offset_col,
                        "type": "quantitative",
                        "scale": {"domain": [-0.5, 0.5]},
                    });
                };
                apply_hinge_offset(&mut lower_hinge);
                apply_hinge_offset(&mut upper_hinge);
            }

            layers.push(lower_hinge);
            layers.push(upper_hinge);
        }

        layers.push(box_part);
        layers.push(median_line);

        Ok(layers)
    }
}

impl GeomRenderer for BoxplotRenderer {
    fn prepare_data(
        &self,
        df: &DataFrame,
        _layer: &Layer,
        _data_key: &str,
        binned_columns: &HashMap<String, Vec<f64>>,
    ) -> Result<PreparedData> {
        let (components, has_outliers) = self.prepare_components(df, binned_columns)?;

        Ok(PreparedData::Composite {
            components,
            metadata: Box::new(BoxplotMetadata { has_outliers }),
        })
    }

    fn needs_source_filter(&self) -> bool {
        // Boxplot uses component-specific filters instead
        false
    }

    fn finalize(
        &self,
        prototype: Value,
        layer: &Layer,
        data_key: &str,
        prepared: &PreparedData,
        context: &RenderContext,
    ) -> Result<Vec<Value>> {
        let PreparedData::Composite { metadata, .. } = prepared else {
            return Err(GgsqlError::InternalError(
                "BoxplotRenderer::finalize called with non-composite data".to_string(),
            ));
        };

        let info = metadata.downcast_ref::<BoxplotMetadata>().ok_or_else(|| {
            GgsqlError::InternalError("Failed to downcast boxplot metadata".to_string())
        })?;

        self.render_layers(prototype, layer, data_key, info.has_outliers, context)
    }
}

// =============================================================================
// Spatial Renderer
// =============================================================================

struct SpatialRenderer;

#[cfg(feature = "spatial")]
impl SpatialRenderer {
    fn wkb_to_geojson(wkb_bytes: &[u8]) -> Result<Value> {
        use geozero::geojson::GeoJsonWriter;
        use geozero::wkb::Wkb;
        use geozero::GeozeroGeometry;
        use std::io::Cursor;

        let mut geojson_out = Vec::new();
        let wkb = Wkb(wkb_bytes);
        wkb.process_geom(&mut GeoJsonWriter::new(Cursor::new(&mut geojson_out)))
            .map_err(|e| {
                GgsqlError::WriterError(format!("Failed to convert WKB to GeoJSON: {}", e))
            })?;

        match serde_json::from_slice(&geojson_out) {
            Ok(value) => Ok(value),
            Err(_) => Ok(Value::Null),
        }
    }

    fn parse_geometry_from_array(array: &arrow::array::ArrayRef, idx: usize) -> Result<Value> {
        use arrow::datatypes::DataType;

        fn decode_hex_wkb(hex: &str) -> Result<Vec<u8>> {
            // PostGIS returns WKB as hex-encoded string via ODBC
            let hex = hex.strip_prefix("\\x").unwrap_or(hex);
            (0..hex.len())
                .step_by(2)
                .map(|i| {
                    u8::from_str_radix(&hex[i..i + 2], 16).map_err(|_| {
                        GgsqlError::WriterError(format!("Invalid hex in WKB at position {}", i))
                    })
                })
                .collect()
        }

        if array.is_null(idx) {
            return Ok(Value::Null);
        }

        match array.data_type() {
            DataType::Binary => {
                let bin = array
                    .as_any()
                    .downcast_ref::<arrow::array::BinaryArray>()
                    .ok_or_else(|| {
                        GgsqlError::WriterError("Failed to read geometry as Binary".into())
                    })?;
                Self::wkb_to_geojson(bin.value(idx))
            }
            DataType::LargeBinary => {
                let bin = array
                    .as_any()
                    .downcast_ref::<arrow::array::LargeBinaryArray>()
                    .ok_or_else(|| {
                        GgsqlError::WriterError("Failed to read geometry as LargeBinary".into())
                    })?;
                Self::wkb_to_geojson(bin.value(idx))
            }
            DataType::Utf8 => {
                let arr = array
                    .as_any()
                    .downcast_ref::<arrow::array::StringArray>()
                    .ok_or_else(|| {
                        GgsqlError::WriterError("Failed to read geometry as Utf8".into())
                    })?;
                let bytes = decode_hex_wkb(arr.value(idx))?;
                Self::wkb_to_geojson(&bytes)
            }
            DataType::LargeUtf8 => {
                let arr = array
                    .as_any()
                    .downcast_ref::<arrow::array::LargeStringArray>()
                    .ok_or_else(|| {
                        GgsqlError::WriterError("Failed to read geometry as LargeUtf8".into())
                    })?;
                let bytes = decode_hex_wkb(arr.value(idx))?;
                Self::wkb_to_geojson(&bytes)
            }
            other => Err(GgsqlError::WriterError(format!(
                "Geometry column has unsupported type {:?}; expected Binary (WKB)",
                other
            ))),
        }
    }
}

impl GeomRenderer for SpatialRenderer {
    fn prepare_data(
        &self,
        df: &DataFrame,
        _layer: &Layer,
        _data_key: &str,
        _binned_columns: &HashMap<String, Vec<f64>>,
    ) -> Result<PreparedData> {
        #[cfg(not(feature = "spatial"))]
        {
            return Err(GgsqlError::WriterError(
                "Spatial visualization requires the 'spatial' feature to be enabled".to_string(),
            ));
        }

        #[cfg(feature = "spatial")]
        {
            let geometry_col = naming::aesthetic_column("geometry");

            let col_names: Vec<String> = df
                .get_column_names()
                .iter()
                .map(|s| s.to_string())
                .collect();

            let mut features = Vec::with_capacity(df.height());

            for row_idx in 0..df.height() {
                let mut feature = serde_json::Map::new();
                feature.insert("type".to_string(), json!("Feature"));

                for col_name in &col_names {
                    let col = df.column(col_name).map_err(|e| {
                        GgsqlError::WriterError(format!(
                            "Failed to get column '{}': {}",
                            col_name, e
                        ))
                    })?;

                    if *col_name == geometry_col {
                        let geom = Self::parse_geometry_from_array(col, row_idx)?;
                        feature.insert("geometry".to_string(), geom);
                    } else if !matches!(
                        // These are reserved names for the geojson format, so
                        // we shouldn't be inserting columns with that name.
                        // Our naming module should prevent such collisions,
                        // so this is purely defensive.
                        col_name.as_str(),
                        "type" | "geometry" | "properties" | "bbox" | "id"
                    ) {
                        let value = super::data::series_value_at(col, row_idx)?;
                        feature.insert(col_name.clone(), value);
                    }
                }
                features.push(Value::Object(feature));
            }

            Ok(PreparedData::Single {
                values: features,
                metadata: Box::new(()),
            })
        }
    }
}

// =============================================================================
// Dispatcher
// =============================================================================

/// Get the appropriate renderer for a geom type
pub fn get_renderer(geom: &Geom) -> Box<dyn GeomRenderer> {
    match geom.geom_type() {
        GeomType::Point => Box::new(PointRenderer),
        GeomType::Path => Box::new(PathRenderer),
        GeomType::Line => Box::new(PathRenderer),
        GeomType::Bar => Box::new(BarRenderer),
        GeomType::Tile => Box::new(TileRenderer),
        GeomType::Polygon => Box::new(PolygonRenderer),
        GeomType::Boxplot => Box::new(BoxplotRenderer),
        GeomType::Violin => Box::new(ViolinRenderer),
        GeomType::Text => Box::new(TextRenderer),
        GeomType::Range => Box::new(RangeRenderer),
        GeomType::Rule => Box::new(RuleRenderer),
        GeomType::Ribbon => Box::new(RibbonRenderer),
        GeomType::Segment => Box::new(SegmentRenderer),
        GeomType::Spatial => Box::new(SpatialRenderer),
        // All other geoms (Point, Area, Density, etc.) use the default renderer
        _ => Box::new(DefaultRenderer),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::plot::Parameters;

    #[test]
    fn test_violin_detail_encoding() {
        let renderer = ViolinRenderer;
        let layer = Layer::new(crate::plot::Geom::violin());

        // Case 1: No detail encoding - should add x
        let mut encoding = serde_json::Map::new();
        encoding.insert(
            "x".to_string(),
            json!({"field": "species", "type": "nominal"}),
        );
        let context = RenderContext::default_for_test();
        renderer
            .modify_encoding(&mut encoding, &layer, &context)
            .unwrap();
        assert_eq!(
            encoding.get("detail"),
            Some(&json!({"field": "species", "type": "nominal"}))
        );

        // Case 2: Detail is single object (not x) - should convert to array
        let mut encoding = serde_json::Map::new();
        encoding.insert(
            "x".to_string(),
            json!({"field": "species", "type": "nominal"}),
        );
        encoding.insert(
            "detail".to_string(),
            json!({"field": "island", "type": "nominal"}),
        );
        let context = RenderContext::default_for_test();
        renderer
            .modify_encoding(&mut encoding, &layer, &context)
            .unwrap();
        assert_eq!(
            encoding.get("detail"),
            Some(&json!([
                {"field": "island", "type": "nominal"},
                {"field": "species", "type": "nominal"}
            ]))
        );

        // Case 3: Detail is single object (already x) - should not change
        let mut encoding = serde_json::Map::new();
        encoding.insert(
            "x".to_string(),
            json!({"field": "species", "type": "nominal"}),
        );
        encoding.insert(
            "detail".to_string(),
            json!({"field": "species", "type": "nominal"}),
        );
        let context = RenderContext::default_for_test();
        renderer
            .modify_encoding(&mut encoding, &layer, &context)
            .unwrap();
        assert_eq!(
            encoding.get("detail"),
            Some(&json!({"field": "species", "type": "nominal"}))
        );

        // Case 4: Detail is array without x - should add x
        let mut encoding = serde_json::Map::new();
        encoding.insert(
            "x".to_string(),
            json!({"field": "species", "type": "nominal"}),
        );
        encoding.insert(
            "detail".to_string(),
            json!([{"field": "island", "type": "nominal"}]),
        );
        let context = RenderContext::default_for_test();
        renderer
            .modify_encoding(&mut encoding, &layer, &context)
            .unwrap();
        assert_eq!(
            encoding.get("detail"),
            Some(&json!([
                {"field": "island", "type": "nominal"},
                {"field": "species", "type": "nominal"}
            ]))
        );

        // Case 5: Detail is array with x already - should not change
        let mut encoding = serde_json::Map::new();
        encoding.insert(
            "x".to_string(),
            json!({"field": "species", "type": "nominal"}),
        );
        encoding.insert(
            "detail".to_string(),
            json!([
                {"field": "island", "type": "nominal"},
                {"field": "species", "type": "nominal"}
            ]),
        );
        let context = RenderContext::default_for_test();
        renderer
            .modify_encoding(&mut encoding, &layer, &context)
            .unwrap();
        assert_eq!(
            encoding.get("detail"),
            Some(&json!([
                {"field": "island", "type": "nominal"},
                {"field": "species", "type": "nominal"}
            ]))
        );
    }

    // =============================================================================
    // TileRenderer Test Helpers
    // =============================================================================

    /// Helper to create a quantitative encoding entry
    fn quant(field: &str) -> Value {
        json!({"field": field, "type": "quantitative"})
    }

    /// Helper to create a nominal encoding entry
    fn nominal(field: &str) -> Value {
        json!({"field": field, "type": "nominal"})
    }

    /// Helper to create a literal value encoding
    fn literal(val: f64) -> Value {
        json!({"value": val})
    }

    /// Helper to create a scale encoding with explicit domain
    /// Use same min/max for constant scales, different values for variable scales
    fn scale(field: &str, min: f64, max: f64) -> Value {
        json!({
            "field": field,
            "type": "quantitative",
            "scale": {
                "domain": [min, max]
            }
        })
    }

    /// Helper to run tile rendering pipeline (modify_encoding + modify_spec)
    fn render_tile(encoding: &mut Map<String, Value>) -> Result<Value> {
        let renderer = TileRenderer;
        let layer = Layer::new(crate::plot::Geom::tile());
        let context = RenderContext::default_for_test();

        renderer.modify_encoding(encoding, &layer, &context)?;

        let mut layer_spec = json!({
            "mark": {"type": "rect", "clip": true},
            "encoding": encoding
        });

        renderer.modify_spec(&mut layer_spec, &layer, &context)?;
        Ok(layer_spec)
    }

    // =============================================================================
    // TileRenderer Tests
    // =============================================================================

    #[test]
    fn test_tile_discrete_x_continuous_y() {
        // Test tile with discrete x scale and continuous y scale
        // x/width (discrete) and y/y2 (continuous, already mapped from pos2min/pos2max)
        let mut encoding = serde_json::Map::new();
        encoding.insert("x".to_string(), nominal("day"));
        encoding.insert("width".to_string(), literal(0.8));
        encoding.insert("y".to_string(), quant("ymin_col"));
        encoding.insert("y2".to_string(), quant("ymax_col"));

        let spec = render_tile(&mut encoding).unwrap();
        let enc = spec["encoding"].as_object().unwrap();

        // x should remain as x (discrete)
        assert_eq!(enc.get("x"), Some(&nominal("day")));

        // y/y2 should be preserved
        assert_eq!(enc.get("y"), Some(&quant("ymin_col")));
        assert_eq!(enc.get("y2"), Some(&quant("ymax_col")));

        // width should be removed from encoding
        assert!(enc.get("width").is_none());

        // Should have mark-level width with band sizing
        assert_eq!(spec["mark"]["width"], json!({"band": 0.8}));
        assert!(spec["mark"].get("height").is_none()); // y is continuous, no height
    }

    #[test]
    fn test_tile_discrete_both_axes_literal_width() {
        // Test tile with discrete scales on both axes with literal width/height
        let mut encoding = serde_json::Map::new();
        encoding.insert("x".to_string(), nominal("day"));
        encoding.insert("width".to_string(), literal(0.7));
        encoding.insert("y".to_string(), nominal("hour"));
        encoding.insert("height".to_string(), literal(0.9));

        let spec = render_tile(&mut encoding).unwrap();
        let enc = spec["encoding"].as_object().unwrap();

        // x and y should remain
        assert_eq!(enc.get("x"), Some(&nominal("day")));
        assert_eq!(enc.get("y"), Some(&nominal("hour")));

        // width/height should be removed from encoding
        assert!(enc.get("width").is_none());
        assert!(enc.get("height").is_none());

        // Should have mark-level width and height with band sizing
        assert_eq!(spec["mark"]["width"], json!({"band": 0.7}));
        assert_eq!(spec["mark"]["height"], json!({"band": 0.9}));
    }

    #[test]
    fn test_tile_discrete_both_axes_default_width() {
        // Test tile with discrete scales on both axes without explicit width/height
        // Should use default band size (1.0)
        let mut encoding = serde_json::Map::new();
        encoding.insert("x".to_string(), nominal("day"));
        encoding.insert("y".to_string(), nominal("hour"));

        let spec = render_tile(&mut encoding).unwrap();

        // With no width/height specified, should have no mark-level width/height
        // (tiles will fill the full band by default)
        assert!(spec["mark"].get("width").is_none());
        assert!(spec["mark"].get("height").is_none());
    }

    #[test]
    fn test_tile_discrete_with_field_width() {
        // Test that field-based width on discrete scales uses datum expressions
        // (works for both variable and constant domains, or no domain)
        let mut encoding = serde_json::Map::new();
        encoding.insert("x".to_string(), nominal("day"));
        encoding.insert("width".to_string(), scale("width_col", 0.5, 0.9));

        let spec = render_tile(&mut encoding).unwrap();

        // Should use mark-level width with datum expression
        assert_eq!(
            spec["mark"]["width"],
            json!({"expr": "datum.width_col * bandwidth('x')"})
        );
    }

    #[test]
    fn test_tile_continuous_x_discrete_y() {
        // Test tile with continuous x (already mapped to x/x2) and discrete y (y/height)
        let mut encoding = serde_json::Map::new();
        encoding.insert("x".to_string(), quant("xmin_col"));
        encoding.insert("x2".to_string(), quant("xmax_col"));
        encoding.insert("y".to_string(), nominal("category"));
        encoding.insert("height".to_string(), literal(0.6));

        let spec = render_tile(&mut encoding).unwrap();
        let enc = spec["encoding"].as_object().unwrap();

        // x/x2 should be preserved
        assert_eq!(enc.get("x"), Some(&quant("xmin_col")));
        assert_eq!(enc.get("x2"), Some(&quant("xmax_col")));

        // y should remain as y (discrete)
        assert_eq!(enc.get("y"), Some(&nominal("category")));

        // height should be removed from encoding
        assert!(enc.get("height").is_none());

        // Should have mark-level height with band sizing
        assert!(spec["mark"].get("width").is_none()); // x is continuous, no width
        assert_eq!(spec["mark"]["height"], json!({"band": 0.6}));
    }
    #[test]
    fn test_text_constant_font() {
        use crate::df;
        use crate::naming;

        let renderer = TextRenderer;
        let layer = Layer::new(crate::plot::Geom::text());

        // Create DataFrame where all rows have the same font
        let x_col = naming::aesthetic_column("x");
        let y_col = naming::aesthetic_column("y");
        let label_col = naming::aesthetic_column("label");
        let typeface_col = naming::aesthetic_column("typeface");
        let df = df! {
            x_col.as_str() => vec![1.0, 2.0, 3.0],
            y_col.as_str() => vec![10.0, 20.0, 30.0],
            label_col.as_str() => vec!["A", "B", "C"],
            typeface_col.as_str() => vec!["Arial", "Arial", "Arial"],
        }
        .unwrap();

        // Prepare data - should result in single layer with _font_0 component key
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        match prepared {
            PreparedData::Composite { components, .. } => {
                // Should have single component with _font_0 key
                assert_eq!(components.len(), 1);
                assert!(components.contains_key("_font_0"));
            }
            _ => panic!("Expected Composite"),
        }
    }

    #[test]
    fn test_text_varying_font() {
        use crate::df;
        use crate::naming;

        let renderer = TextRenderer;
        let layer = Layer::new(crate::plot::Geom::text());

        // Create DataFrame with different fonts per row
        let x_col = naming::aesthetic_column("x");
        let y_col = naming::aesthetic_column("y");
        let label_col = naming::aesthetic_column("label");
        let typeface_col = naming::aesthetic_column("typeface");
        let df = df! {
            x_col.as_str() => vec![1.0, 2.0, 3.0],
            y_col.as_str() => vec![10.0, 20.0, 30.0],
            label_col.as_str() => vec!["A", "B", "C"],
            typeface_col.as_str() => vec!["Arial", "Courier", "Times"],
        }
        .unwrap();

        // Prepare data - should result in multiple layers
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        match prepared {
            PreparedData::Composite { components, .. } => {
                // Should have 3 components (one per unique font) with suffix keys
                assert_eq!(components.len(), 3);
                assert!(components.contains_key("_font_0"));
                assert!(components.contains_key("_font_1"));
                assert!(components.contains_key("_font_2"));
            }
            _ => panic!("Expected Composite"),
        }
    }

    #[test]
    fn test_text_nested_layers_structure() {
        use crate::df;
        use crate::naming;

        let renderer = TextRenderer;
        let layer = Layer::new(crate::plot::Geom::text());

        // Create DataFrame with different fonts
        let x_col = naming::aesthetic_column("x");
        let y_col = naming::aesthetic_column("y");
        let label_col = naming::aesthetic_column("label");
        let typeface_col = naming::aesthetic_column("typeface");
        let fontweight_col = naming::aesthetic_column("fontweight");
        let italic_col = naming::aesthetic_column("italic");
        let df = df! {
            x_col.as_str() => vec![1.0, 2.0, 3.0],
            y_col.as_str() => vec![10.0, 20.0, 30.0],
            label_col.as_str() => vec!["A", "B", "C"],
            typeface_col.as_str() => vec!["Arial", "Courier", "Arial"],
            fontweight_col.as_str() => vec!["bold", "normal", "bold"],
            italic_col.as_str() => vec!["false", "true", "false"],
        }
        .unwrap();

        // Prepare data
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        // Get the components
        let components = match &prepared {
            PreparedData::Composite { components, .. } => components,
            _ => panic!("Expected Composite"),
        };

        // Should have 3 components due to non-contiguous indices
        // (Arial+bold+not-italic at index 0, Courier+normal+italic at index 1, Arial+bold+not-italic at index 2)
        assert_eq!(components.len(), 3);

        // Build prototype spec
        let prototype = json!({
            "mark": {"type": "text"},
            "encoding": {
                "x": {"field": naming::aesthetic_column("x"), "type": "quantitative"},
                "y": {"field": naming::aesthetic_column("y"), "type": "quantitative"},
                "text": {"field": naming::aesthetic_column("label"), "type": "nominal"}
            }
        });

        // Create a dummy layer
        let layer = crate::plot::Layer::new(crate::plot::Geom::text());
        let context = RenderContext::default_for_test();

        // Call finalize to get layers
        let layers = renderer
            .finalize(prototype.clone(), &layer, "test", &prepared, &context)
            .unwrap();

        // For multiple font groups, should return single parent spec with nested layers
        assert_eq!(layers.len(), 1);

        let parent_spec = &layers[0];

        // Parent should have "layer" array
        assert!(parent_spec.get("layer").is_some());
        let nested_layers = parent_spec["layer"].as_array().unwrap();

        // Should have 3 nested layers (one per component)
        assert_eq!(nested_layers.len(), 3);

        // Parent should have shared encoding
        assert!(parent_spec.get("encoding").is_some());

        // Each nested layer should have mark and transform, but not encoding
        for nested_layer in nested_layers {
            assert!(nested_layer.get("mark").is_some());
            assert!(nested_layer.get("transform").is_some());
            assert!(nested_layer.get("encoding").is_none());

            // Mark should have font properties
            let mark = nested_layer["mark"].as_object().unwrap();
            assert!(mark.contains_key("fontWeight"));
            assert!(mark.contains_key("fontStyle"));
        }
    }

    #[test]
    fn test_text_varying_angle() {
        use crate::df;
        use crate::naming;

        let renderer = TextRenderer;
        let layer = Layer::new(crate::plot::Geom::text());

        // Create DataFrame with different angles
        let x_col = naming::aesthetic_column("x");
        let y_col = naming::aesthetic_column("y");
        let label_col = naming::aesthetic_column("label");
        let rotation_col = naming::aesthetic_column("rotation");
        let df = df! {
            x_col.as_str() => vec![1.0, 2.0, 3.0],
            y_col.as_str() => vec![10.0, 20.0, 30.0],
            label_col.as_str() => vec!["A", "B", "C"],
            rotation_col.as_str() => vec!["0", "45", "90"],
        }
        .unwrap();

        // Prepare data - should result in multiple layers (one per unique angle)
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        match &prepared {
            PreparedData::Composite { components, .. } => {
                // Should have 3 components (one per unique angle)
                assert_eq!(components.len(), 3);
                assert!(components.contains_key("_font_0"));
                assert!(components.contains_key("_font_1"));
                assert!(components.contains_key("_font_2"));
            }
            _ => panic!("Expected Composite"),
        }

        // Build prototype spec
        let prototype = json!({
            "mark": {"type": "text"},
            "encoding": {
                "x": {"field": naming::aesthetic_column("x"), "type": "quantitative"},
                "y": {"field": naming::aesthetic_column("y"), "type": "quantitative"},
                "text": {"field": naming::aesthetic_column("label"), "type": "nominal"}
            }
        });

        // Create a dummy layer
        let layer = crate::plot::Layer::new(crate::plot::Geom::text());
        let context = RenderContext::default_for_test();

        // Call finalize to get layers
        let layers = renderer
            .finalize(prototype.clone(), &layer, "test", &prepared, &context)
            .unwrap();

        // Should return single parent spec with nested layers
        assert_eq!(layers.len(), 1);

        let parent_spec = &layers[0];
        let nested_layers = parent_spec["layer"].as_array().unwrap();

        // Should have 3 nested layers (one per unique angle)
        assert_eq!(nested_layers.len(), 3);

        // Each layer should have angle property in mark
        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert!(mark.contains_key("angle")); // Vega-Lite uses "angle" property name
        }
    }

    #[test]
    fn test_text_varying_angle_numeric() {
        use crate::df;
        use crate::naming;

        let renderer = TextRenderer;
        let layer = Layer::new(crate::plot::Geom::text());

        // Create DataFrame with numeric angle column (matching actual query)
        let x_col = naming::aesthetic_column("x");
        let y_col = naming::aesthetic_column("y");
        let label_col = naming::aesthetic_column("label");
        let rotation_col = naming::aesthetic_column("rotation");
        let df = df! {
            x_col.as_str() => vec![1i32, 2, 3],
            y_col.as_str() => vec![1i32, 2, 3],
            label_col.as_str() => vec!["A", "B", "C"],
            rotation_col.as_str() => vec![0i32, 180i32, 0i32],  // integer column
        }
        .unwrap();

        // Prepare data - should result in multiple layers (one per unique angle)
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        match &prepared {
            PreparedData::Composite { components, .. } => {
                // Should have 3 components: angle 0 at row 0, angle 180 at row 1, angle 0 at row 2
                // Due to non-contiguous indices, rows 0 and 2 should be in separate components
                eprintln!("Number of components: {}", components.len());
                eprintln!(
                    "Component keys: {:?}",
                    components.keys().collect::<Vec<_>>()
                );
                assert_eq!(components.len(), 3);
            }
            _ => panic!("Expected Composite"),
        }
    }

    #[test]
    fn test_text_angle_integration() {
        use crate::execute;
        use crate::naming;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Integration test: Full pipeline from SQL query to Vega-Lite with angle aesthetic
        // This tests that angle values properly create separate layers with angle mark properties

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        // Query with text geom and varying angles
        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                chr(65 + n::INTEGER) as label,
                CASE
                    WHEN n = 0 THEN 0
                    WHEN n = 1 THEN 45
                    WHEN n = 2 THEN 90
                    ELSE 0
                END as rot
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, label, rot AS rotation
            DRAW text
        "#;

        // Execute and prepare data
        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        assert_eq!(prepared.specs.len(), 1);

        let spec = &prepared.specs[0];
        assert_eq!(spec.layers.len(), 1);

        // Generate Vega-Lite JSON
        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        // Text renderer should create nested layers structure
        assert!(
            vl_spec["layer"].is_array(),
            "Should have top-level layer array"
        );
        let top_layers = vl_spec["layer"].as_array().unwrap();
        assert_eq!(top_layers.len(), 1, "Should have one parent text layer");

        // Parent layer should have shared encoding and nested layers
        let parent_layer = &top_layers[0];
        assert!(
            parent_layer["encoding"].is_object(),
            "Parent layer should have shared encoding"
        );
        assert!(
            parent_layer["layer"].is_array(),
            "Parent layer should have nested layers"
        );

        let nested_layers = parent_layer["layer"].as_array().unwrap();

        // Should have multiple nested layers (one per unique angle value)
        // We have angles: 0, 45, 90, 0 -> but non-contiguous 0s split into separate layers
        assert!(
            nested_layers.len() >= 3,
            "Should have at least 3 nested layers for different angles, got {}",
            nested_layers.len()
        );

        // Each nested layer should have mark with angle property
        for (idx, nested_layer) in nested_layers.iter().enumerate() {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert!(
                mark.contains_key("angle"), // Vega-Lite uses "angle" property name
                "Nested layer {} mark should have angle property",
                idx
            );
            assert_eq!(mark["type"], "text");

            // Should have source filter transform
            assert!(nested_layer["transform"].is_array());

            // Should NOT have encoding (inherited from parent)
            assert!(nested_layer.get("encoding").is_none());
        }

        // Verify angles are present and normalized [0, 360)
        let angles: Vec<f64> = nested_layers
            .iter()
            .filter_map(|layer| {
                layer["mark"]
                    .as_object()
                    .and_then(|m| m.get("angle"))
                    .and_then(|a| a.as_f64())
            })
            .collect();

        // Should have the three distinct angles: 0, 45, 90
        assert!(angles.contains(&0.0), "Should have 0° angle");
        assert!(angles.contains(&45.0), "Should have 45° angle");
        assert!(angles.contains(&90.0), "Should have 90° angle");

        // Verify data has angle column
        let data_values = vl_spec["data"]["values"].as_array().unwrap();
        assert!(!data_values.is_empty());

        let angle_col = naming::aesthetic_column("rotation");
        for row in data_values {
            assert!(
                row[&angle_col].is_number(),
                "Data row should have numeric angle: {:?}",
                row
            );
        }
    }

    #[test]
    fn test_text_offset_parameters() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Integration test: offset parameter should map to xOffset/yOffset

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        // Query with offset parameter
        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                chr(65 + n::INTEGER) as label
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, label
            DRAW text SETTING offset => [5, -10]
        "#;

        // Execute and prepare data
        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        assert_eq!(prepared.specs.len(), 1);

        let spec = &prepared.specs[0];
        assert_eq!(spec.layers.len(), 1);

        // Generate Vega-Lite JSON
        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        // Text renderer creates nested layers structure
        let top_layers = vl_spec["layer"].as_array().unwrap();
        assert_eq!(top_layers.len(), 1);

        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        // All nested layers should have xOffset and yOffset in mark
        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();

            assert!(
                mark.contains_key("xOffset"),
                "Mark should have xOffset from offset"
            );
            assert_eq!(
                mark["xOffset"].as_f64().unwrap(),
                5.0 * POINTS_TO_PIXELS,
                "xOffset should be 5 * POINTS_TO_PIXELS"
            );

            assert!(
                mark.contains_key("yOffset"),
                "Mark should have yOffset from offset"
            );
            assert_eq!(
                mark["yOffset"].as_f64().unwrap(),
                10.0 * POINTS_TO_PIXELS,
                "yOffset should be 10 * POINTS_TO_PIXELS (negated from offset[1] = -10)"
            );
        }
    }

    #[test]
    fn test_text_label_formatting() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Integration test: format parameter should transform label values

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        // Query with format parameter using Title case transformation
        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                CASE
                    WHEN n = 0 THEN 'north region'
                    WHEN n = 1 THEN 'south region'
                    ELSE 'east region'
                END as region
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, region AS label
            DRAW text SETTING format => 'Region: {:Title}'
        "#;

        // Execute and prepare data
        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        assert_eq!(prepared.specs.len(), 1);

        let spec = &prepared.specs[0];
        assert_eq!(spec.layers.len(), 1);

        // Generate Vega-Lite JSON
        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        // Check that data has formatted labels
        let data_values = vl_spec["data"]["values"].as_array().unwrap();
        assert!(!data_values.is_empty());

        // Verify formatted labels in the data
        let label_col = crate::naming::aesthetic_column("label");

        // Check each row has properly formatted labels
        let labels: Vec<&str> = data_values
            .iter()
            .filter_map(|row| row[&label_col].as_str())
            .collect();

        assert_eq!(labels.len(), 3);
        assert!(labels.contains(&"Region: North Region"));
        assert!(labels.contains(&"Region: South Region"));
        assert!(labels.contains(&"Region: East Region"));
    }

    #[test]
    fn test_text_label_formatting_numeric() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Test numeric formatting with printf-style format

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                n::FLOAT * 10.5 as value
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, value AS label
            DRAW text SETTING format => '${:num %.2f}'
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let data_values = vl_spec["data"]["values"].as_array().unwrap();
        let label_col = crate::naming::aesthetic_column("label");

        let labels: Vec<&str> = data_values
            .iter()
            .filter_map(|row| row[&label_col].as_str())
            .collect();

        // Should have formatted currency values
        assert_eq!(labels.len(), 3);
        assert!(labels.contains(&"$0.00"));
        assert!(labels.contains(&"$10.50"));
        assert!(labels.contains(&"$21.00"));
    }

    #[test]
    fn test_text_label_newline_splitting() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Test that labels containing newlines are split into arrays

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        // Query with labels containing newlines in DRAW and PLACE
        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                CASE
                    WHEN n = 0 THEN 'First Line\nSecond Line'
                    WHEN n = 1 THEN 'Single Line'
                    ELSE 'Line 1\nLine 2\nLine 3'
                END as label
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, label
            DRAW text
            PLACE text SETTING x => 5, y => 15, label => 'Annotation\nWith Newline'
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let data_values = vl_spec["data"]["values"].as_array().unwrap();
        let label_col = crate::naming::aesthetic_column("label");

        // Check first label (contains newline - should be array)
        let label_0 = &data_values[0][&label_col];
        assert!(label_0.is_array(), "Label with newline should be an array");
        let lines_0 = label_0.as_array().unwrap();
        assert_eq!(lines_0.len(), 2);
        assert_eq!(lines_0[0].as_str().unwrap(), "First Line");
        assert_eq!(lines_0[1].as_str().unwrap(), "Second Line");

        // Check second label (no newline - should be string)
        let label_1 = &data_values[1][&label_col];
        assert!(
            label_1.is_string(),
            "Label without newline should be a string"
        );
        assert_eq!(label_1.as_str().unwrap(), "Single Line");

        // Check third label (multiple newlines - should be array with 3 elements)
        let label_2 = &data_values[2][&label_col];
        assert!(label_2.is_array(), "Label with newlines should be an array");
        let lines_2 = label_2.as_array().unwrap();
        assert_eq!(lines_2.len(), 3);
        assert_eq!(lines_2[0].as_str().unwrap(), "Line 1");
        assert_eq!(lines_2[1].as_str().unwrap(), "Line 2");
        assert_eq!(lines_2[2].as_str().unwrap(), "Line 3");

        // Check PLACE annotation layer (index 3, after the 3 DRAW data rows)
        assert!(data_values.len() > 3, "Should have annotation data");
        let annotation_label = &data_values[3][&label_col];
        assert!(
            annotation_label.is_array(),
            "Annotation label with newline should be an array"
        );
        let annotation_lines = annotation_label.as_array().unwrap();
        assert_eq!(annotation_lines.len(), 2);
        assert_eq!(annotation_lines[0].as_str().unwrap(), "Annotation");
        assert_eq!(annotation_lines[1].as_str().unwrap(), "With Newline");
    }

    #[test]
    fn test_text_setting_fontweight() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Integration test: SETTING fontweight => 'bold' should add fontWeight to base mark

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        // Query with fontweight in SETTING
        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                chr(65 + n::INTEGER) as label
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, label
            DRAW text SETTING fontweight => 'bold'
        "#;

        // Execute and prepare data
        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        assert_eq!(prepared.specs.len(), 1);

        let spec = &prepared.specs[0];
        assert_eq!(spec.layers.len(), 1);

        // Generate Vega-Lite JSON
        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        // Text renderer creates nested layers structure
        let top_layers = vl_spec["layer"].as_array().unwrap();
        assert_eq!(top_layers.len(), 1);

        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        // All nested layers should have fontWeight: "bold" in mark (from SETTING)
        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();

            assert!(
                mark.contains_key("fontWeight"),
                "Mark should have fontWeight from SETTING fontweight"
            );
            assert_eq!(
                mark["fontWeight"].as_str().unwrap(),
                "bold",
                "fontWeight should be bold"
            );
        }
    }

    #[test]
    fn test_text_setting_fontweight_numeric() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Test numeric fontweight values (700 should map to 'bold')
        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                chr(65 + n::INTEGER) as label
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, label
            DRAW text SETTING fontweight => 700
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let top_layers = vl_spec["layer"].as_array().unwrap();
        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        // Numeric 700 should map to 'bold'
        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert_eq!(mark["fontWeight"].as_str().unwrap(), "bold");
        }
    }

    #[test]
    fn test_text_setting_fontweight_numeric_normal() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        // Test numeric fontweight values (400 should map to 'normal')
        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                chr(65 + n::INTEGER) as label
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, label
            DRAW text SETTING fontweight => 400
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let top_layers = vl_spec["layer"].as_array().unwrap();
        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        // Numeric 400 should map to 'normal'
        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert_eq!(mark["fontWeight"].as_str().unwrap(), "normal");
        }
    }

    #[test]
    fn test_text_setting_fontweight_keywords() {
        use crate::execute;
        use crate::reader::DuckDBReader;
        use crate::writer::vegalite::VegaLiteWriter;
        use crate::writer::Writer;

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();

        // Test 'bolder' keyword (should map to 'bold')
        let query = r#"
            SELECT 1 as x, 1 as y, 'A' as label
            VISUALISE x, y, label
            DRAW text SETTING fontweight => 'bolder'
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let writer = VegaLiteWriter::new();
        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let top_layers = vl_spec["layer"].as_array().unwrap();
        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert_eq!(mark["fontWeight"].as_str().unwrap(), "bold");
        }

        // Test 'lighter' keyword (should map to 'normal')
        let query = r#"
            SELECT 1 as x, 1 as y, 'A' as label
            VISUALISE x, y, label
            DRAW text SETTING fontweight => 'lighter'
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let top_layers = vl_spec["layer"].as_array().unwrap();
        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert_eq!(mark["fontWeight"].as_str().unwrap(), "normal");
        }

        // Test 'semi-bold' keyword (should map to 'bold' since 600 >= 500)
        let query = r#"
            SELECT 1 as x, 1 as y, 'A' as label
            VISUALISE x, y, label
            DRAW text SETTING fontweight => 'semi-bold'
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let top_layers = vl_spec["layer"].as_array().unwrap();
        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert_eq!(mark["fontWeight"].as_str().unwrap(), "bold");
        }

        // Test 'light' keyword (should map to 'normal' since 300 < 500)
        let query = r#"
            SELECT 1 as x, 1 as y, 'A' as label
            VISUALISE x, y, label
            DRAW text SETTING fontweight => 'light'
        "#;

        let prepared = execute::prepare_data_with_reader(query, &reader).unwrap();
        let spec = &prepared.specs[0];

        let json_str = writer.write(spec, &prepared.data).unwrap();
        let vl_spec: serde_json::Value = serde_json::from_str(&json_str).unwrap();

        let top_layers = vl_spec["layer"].as_array().unwrap();
        let parent_layer = &top_layers[0];
        let nested_layers = parent_layer["layer"].as_array().unwrap();

        for nested_layer in nested_layers {
            let mark = nested_layer["mark"].as_object().unwrap();
            assert_eq!(mark["fontWeight"].as_str().unwrap(), "normal");
        }
    }

    #[test]
    fn test_fontweight_keyword_to_numeric_conversion() {
        // Test parse_fontweight_to_numeric helper function - all CSS keywords

        // 100 - thin/hairline
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("thin"),
            Some(100.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("hairline"),
            Some(100.0)
        );

        // 200 - extra-light/ultra-light
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("extra-light"),
            Some(200.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("extralight"),
            Some(200.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("ultra-light"),
            Some(200.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("ultralight"),
            Some(200.0)
        );

        // 300 - light
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("light"),
            Some(300.0)
        );

        // 400 - normal/regular/lighter
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("normal"),
            Some(400.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("regular"),
            Some(400.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("lighter"),
            Some(400.0)
        );

        // 500 - medium
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("medium"),
            Some(500.0)
        );

        // 600 - semi-bold/demi-bold
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("semi-bold"),
            Some(600.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("semibold"),
            Some(600.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("demi-bold"),
            Some(600.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("demibold"),
            Some(600.0)
        );

        // 700 - bold/bolder
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("bold"),
            Some(700.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("bolder"),
            Some(700.0)
        );

        // 800 - extra-bold/ultra-bold
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("extra-bold"),
            Some(800.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("extrabold"),
            Some(800.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("ultra-bold"),
            Some(800.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("ultrabold"),
            Some(800.0)
        );

        // 900 - black/heavy
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("black"),
            Some(900.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("heavy"),
            Some(900.0)
        );

        // Case insensitive
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("BOLD"),
            Some(700.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("Normal"),
            Some(400.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("SEMI-BOLD"),
            Some(600.0)
        );

        // Numeric strings pass through
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("100"),
            Some(100.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("400"),
            Some(400.0)
        );
        assert_eq!(
            TextRenderer::parse_fontweight_to_numeric("700"),
            Some(700.0)
        );

        // Invalid values
        assert_eq!(TextRenderer::parse_fontweight_to_numeric("invalid"), None);
        assert_eq!(TextRenderer::parse_fontweight_to_numeric(""), None);
    }

    #[test]
    fn test_violin_mirroring() {
        use crate::naming;

        let renderer = ViolinRenderer;
        let context = RenderContext::default_for_test();

        let layer = Layer::new(crate::plot::Geom::violin());
        let mut layer_spec = json!({
            "mark": {"type": "line"},
            "encoding": {
                "x": {"field": "species", "type": "nominal"},
                "y": {"field": naming::aesthetic_column("pos2"), "type": "quantitative"}
            }
        });

        renderer
            .modify_spec(&mut layer_spec, &layer, &context)
            .unwrap();

        // Verify transforms include mirroring (violin_offsets)
        let transforms = layer_spec["transform"].as_array().unwrap();

        // Find the violin_offsets calculation (mirrors offset on both sides)
        let mirror_calc = transforms
            .iter()
            .find(|t| t.get("as").and_then(|a| a.as_str()) == Some("violin_offsets"));
        assert!(
            mirror_calc.is_some(),
            "Should have violin_offsets mirroring calculation"
        );

        let calc_expr = mirror_calc.unwrap()["calculate"].as_str().unwrap();
        let offset_col = naming::aesthetic_column("offset");
        // Should mirror the offset column: [datum.offset, -datum.offset]
        assert!(
            calc_expr.contains(&offset_col),
            "Mirror calculation should use offset column: {}",
            calc_expr
        );
        assert!(
            calc_expr.contains("-datum"),
            "Mirror calculation should negate: {}",
            calc_expr
        );

        // Verify flatten transform exists
        let flatten = transforms.iter().find(|t| t.get("flatten").is_some());
        assert!(
            flatten.is_some(),
            "Should have flatten transform for violin_offsets"
        );

        // Verify __final_offset calculation (combines with dodge offset)
        let final_offset = transforms
            .iter()
            .find(|t| t.get("as").and_then(|a| a.as_str()) == Some("__final_offset"));
        assert!(
            final_offset.is_some(),
            "Should have __final_offset calculation"
        );
    }

    #[test]
    fn test_violin_ridge_parameter() {
        use crate::naming;
        use crate::plot::ParameterValue;

        let offset_col = naming::aesthetic_column("offset");

        fn get_violin_offset_expr(ridge: Option<&str>, is_horizontal: bool) -> String {
            let mut layer = Layer::new(crate::plot::Geom::violin());
            if let Some(r) = ridge {
                layer
                    .parameters
                    .insert("side".to_string(), ParameterValue::String(r.to_string()));
            }

            // Set orientation parameter for horizontal case
            if is_horizontal {
                layer.parameters.insert(
                    "orientation".to_string(),
                    ParameterValue::String("transposed".to_string()),
                );
            }

            let mut layer_spec = if is_horizontal {
                json!({
                    "mark": {"type": "line"},
                    "encoding": {
                        "x": {"field": naming::aesthetic_column("pos2"), "type": "quantitative"},
                        "y": {"field": "species", "type": "nominal"}
                    }
                })
            } else {
                json!({
                    "mark": {"type": "line"},
                    "encoding": {
                        "x": {"field": "species", "type": "nominal"},
                        "y": {"field": naming::aesthetic_column("pos2"), "type": "quantitative"}
                    }
                })
            };

            ViolinRenderer
                .modify_spec(&mut layer_spec, &layer, &RenderContext::default_for_test())
                .unwrap();

            layer_spec["transform"]
                .as_array()
                .unwrap()
                .iter()
                .find(|t| t.get("as").and_then(|a| a.as_str()) == Some("violin_offsets"))
                .unwrap()["calculate"]
                .as_str()
                .unwrap()
                .to_string()
        }

        // Default "both" - mirrors on both sides (vertical orientation)
        let expr = get_violin_offset_expr(None, false);
        assert!(
            expr.contains(&format!("[datum.{}, -datum.{}]", offset_col, offset_col))
                || expr.contains(&format!("[-datum.{}, datum.{}]", offset_col, offset_col)),
            "Default should mirror both sides: {}",
            expr
        );

        // Explicit "both" - mirrors on both sides (vertical orientation)
        let expr = get_violin_offset_expr(Some("both"), false);
        assert!(
            expr.contains(&format!("[datum.{}, -datum.{}]", offset_col, offset_col))
                || expr.contains(&format!("[-datum.{}, datum.{}]", offset_col, offset_col)),
            "Explicit 'both' should mirror both sides (vertical): {}",
            expr
        );

        // Explicit "both" - mirrors on both sides (horizontal orientation)
        let expr = get_violin_offset_expr(Some("both"), true);
        assert!(
            expr.contains(&format!("[datum.{}, -datum.{}]", offset_col, offset_col))
                || expr.contains(&format!("[-datum.{}, datum.{}]", offset_col, offset_col)),
            "Explicit 'both' should mirror both sides (horizontal): {}",
            expr
        );

        // Vertical orientation (default): x=nominal, y=quantitative
        // "left" and "bottom" - only negative offset
        assert_eq!(
            get_violin_offset_expr(Some("left"), false),
            format!("[-datum.{}]", offset_col)
        );
        assert_eq!(
            get_violin_offset_expr(Some("bottom"), false),
            format!("[-datum.{}]", offset_col)
        );

        // "right" and "top" - only positive offset
        assert_eq!(
            get_violin_offset_expr(Some("right"), false),
            format!("[datum.{}]", offset_col)
        );
        assert_eq!(
            get_violin_offset_expr(Some("top"), false),
            format!("[datum.{}]", offset_col)
        );

        // Horizontal orientation: x=quantitative, y=nominal
        // "bottom" and "left" - only positive offset
        assert_eq!(
            get_violin_offset_expr(Some("bottom"), true),
            format!("[datum.{}]", offset_col)
        );
        assert_eq!(
            get_violin_offset_expr(Some("left"), true),
            format!("[datum.{}]", offset_col)
        );

        // "top" and "right" - only negative offset
        assert_eq!(
            get_violin_offset_expr(Some("top"), true),
            format!("[-datum.{}]", offset_col)
        );
        assert_eq!(
            get_violin_offset_expr(Some("right"), true),
            format!("[-datum.{}]", offset_col)
        );
    }

    #[test]
    fn test_boxplot_side_parameter() {
        use crate::plot::ParameterValue;
        use crate::AestheticValue;

        // Render the boxplot layers with the given `side` and orientation,
        // then return the box, median, lower-whisker, upper-whisker, and
        // outlier mark specs as a tuple for assertion.
        fn render_marks(
            side: Option<&str>,
            is_horizontal: bool,
        ) -> (Value, Value, Value, Value, Value) {
            let mut layer = Layer::new(crate::plot::Geom::boxplot());
            layer
                .mappings
                .insert("pos1", AestheticValue::standard_column("species"));
            layer
                .mappings
                .insert("pos2", AestheticValue::standard_column("bill_len"));
            if let Some(s) = side {
                layer
                    .parameters
                    .insert("side".to_string(), ParameterValue::String(s.to_string()));
            }
            if is_horizontal {
                layer.parameters.insert(
                    "orientation".to_string(),
                    ParameterValue::String("transposed".to_string()),
                );
            }

            // Build a minimal prototype with encoding fields the renderer reads.
            let prototype = if is_horizontal {
                json!({
                    "mark": {"type": "point"},
                    "encoding": {
                        "x": {"field": naming::aesthetic_column("pos1"), "type": "quantitative"},
                        "y": {"field": "species", "type": "nominal"},
                    },
                })
            } else {
                json!({
                    "mark": {"type": "point"},
                    "encoding": {
                        "x": {"field": "species", "type": "nominal"},
                        "y": {"field": naming::aesthetic_column("pos2"), "type": "quantitative"},
                    },
                })
            };

            let layers = BoxplotRenderer
                .render_layers(
                    prototype,
                    &layer,
                    "__ggsql_layer_0__",
                    true, // include outliers
                    &RenderContext::default_for_test(),
                )
                .unwrap();

            // Order set by render_layers: outlier (if has_outliers), lower_whisker,
            // upper_whisker, box, median.
            (
                layers[3].clone(), // box
                layers[4].clone(), // median
                layers[1].clone(), // lower_whisker
                layers[2].clone(), // upper_whisker
                layers[0].clone(), // outlier
            )
        }

        // Helper: extract the band-fraction shift from the offset encoding's
        // calculate transform. Returns 0.0 if no side shift transform is present.
        fn extract_side_shift(layer_spec: &Value) -> f64 {
            let transforms = match layer_spec.get("transform").and_then(|t| t.as_array()) {
                Some(t) => t,
                None => return 0.0,
            };
            for t in transforms {
                if let Some(as_field) = t.get("as").and_then(|s| s.as_str()) {
                    if as_field == "__ggsql_box_side_offset__" {
                        let calc = t.get("calculate").and_then(|c| c.as_str()).unwrap_or("");
                        let after = match calc.split(") + ").nth(1) {
                            Some(s) => s,
                            None => return 0.0,
                        };
                        return after.parse::<f64>().unwrap_or(0.0);
                    }
                }
            }
            0.0
        }

        // Default width=0.9 → side shift magnitude = 0.225 (= w/4).
        let shift_mag = 0.225;

        // Default ("both") in vertical orientation: full width, no side shift.
        let (box_v, _median_v, low_v, up_v, out_v) = render_marks(None, false);
        assert_eq!(extract_side_shift(&box_v), 0.0);
        // Box width should be the full-width expression (not halved).
        let box_full_width = box_v["mark"]["width"].clone();
        assert!(
            box_full_width
                .get("expr")
                .and_then(|e| e.as_str())
                .map(|s| !s.contains("/ 2"))
                .unwrap_or(false),
            "default box width should be full bandwidth*w, got {}",
            box_full_width
        );

        // Vertical "right" / "top" → positive shift (box on right half of band).
        for s in ["right", "top"] {
            let (b, m, low, up, out) = render_marks(Some(s), false);
            let shift = extract_side_shift(&b);
            assert!(
                (shift - shift_mag).abs() < 1e-9,
                "side={s}: expected shift +{shift_mag}, got {shift}"
            );
            assert!(
                (extract_side_shift(&m) - shift_mag).abs() < 1e-9,
                "side={s} median shift mismatch"
            );
            // Box width is halved.
            let bw = b["mark"]["width"]["expr"].as_str().unwrap_or("");
            assert!(
                bw.contains("/ 2"),
                "side={s} expected halved width, got {bw}"
            );
            // xOffset encoding points at the combined column.
            assert_eq!(
                b["encoding"]["xOffset"]["field"],
                json!("__ggsql_box_side_offset__"),
                "side={s}"
            );
            // Whiskers and outliers unchanged.
            assert_eq!(low["mark"], low_v["mark"], "lower whisker side={s}");
            assert_eq!(up["mark"], up_v["mark"], "upper whisker side={s}");
            assert_eq!(out["mark"], out_v["mark"], "outlier side={s}");
            assert!(
                low.get("transform")
                    .and_then(|t| t.as_array())
                    .map(
                        |arr| arr.iter().all(|tr| tr.get("as").and_then(|s| s.as_str())
                            != Some("__ggsql_box_side_offset__"))
                    )
                    .unwrap_or(true),
                "side={s} lower whisker should not have side-shift transform"
            );
        }

        // Vertical "left" / "bottom" → negative shift (box on left half of band).
        for s in ["left", "bottom"] {
            let (b, m, _, _, _) = render_marks(Some(s), false);
            let shift = extract_side_shift(&b);
            assert!(
                (shift + shift_mag).abs() < 1e-9,
                "side={s}: expected shift -{shift_mag}, got {shift}"
            );
            assert!(
                (extract_side_shift(&m) + shift_mag).abs() < 1e-9,
                "side={s} median shift mismatch"
            );
        }

        // Horizontal "both": full height, no shift.
        let (box_h, _, _, _, _) = render_marks(None, true);
        assert_eq!(extract_side_shift(&box_h), 0.0);

        // Horizontal "bottom" / "left" → positive (per violin convention,
        // mapped to positive yOffset which renders below the centerline).
        for s in ["bottom", "left"] {
            let (b, m, _, _, _) = render_marks(Some(s), true);
            assert!(
                (extract_side_shift(&b) - shift_mag).abs() < 1e-9,
                "side={s}: expected +{shift_mag}"
            );
            assert!(
                (extract_side_shift(&m) - shift_mag).abs() < 1e-9,
                "side={s} median"
            );
            // Horizontal uses yOffset, not xOffset.
            assert_eq!(
                b["encoding"]["yOffset"]["field"],
                json!("__ggsql_box_side_offset__"),
                "side={s}"
            );
        }

        // Horizontal "top" / "right" → negative.
        for s in ["top", "right"] {
            let (b, m, _, _, _) = render_marks(Some(s), true);
            assert!(
                (extract_side_shift(&b) + shift_mag).abs() < 1e-9,
                "side={s}: expected -{shift_mag}"
            );
            assert!(
                (extract_side_shift(&m) + shift_mag).abs() < 1e-9,
                "side={s} median"
            );
        }
    }

    #[test]
    fn test_render_context_get_extent() {
        use crate::plot::{ArrayElement, Scale};
        use crate::writer::vegalite::projection::get_projection_renderer;

        let cartesian = get_projection_renderer(None, None, &[]);

        // Test success case: continuous scale with numeric range
        let scales = vec![Scale {
            aesthetic: "x".to_string(),
            scale_type: None,
            input_range: Some(vec![ArrayElement::Number(0.0), ArrayElement::Number(10.0)]),
            explicit_input_range: false,
            output_range: None,
            transform: None,
            explicit_transform: false,
            properties: Parameters::new(),
            resolved: false,
            label_mapping: None,
            label_template: "{}".to_string(),
        }];
        let context = RenderContext::new(
            &scales,
            cartesian.as_ref(),
            AestheticContext::from_static(&["x", "y"], &[]),
        );
        let result = context.get_extent("x");
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), (0.0, 10.0));

        // Test error case: scale not found
        let context = RenderContext::new(
            &scales,
            cartesian.as_ref(),
            AestheticContext::from_static(&["x", "y"], &[]),
        );
        let result = context.get_extent("y");
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("no scale found"));

        // Test error case: scale with no range
        let scales = vec![Scale {
            aesthetic: "x".to_string(),
            scale_type: None,
            input_range: None,
            explicit_input_range: false,
            output_range: None,
            transform: None,
            explicit_transform: false,
            properties: Parameters::new(),
            resolved: false,
            label_mapping: None,
            label_template: "{}".to_string(),
        }];
        let context = RenderContext::new(
            &scales,
            cartesian.as_ref(),
            AestheticContext::from_static(&["x", "y"], &[]),
        );
        let result = context.get_extent("x");
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("no valid numeric range"));

        // Test error case: scale with non-numeric range
        let scales = vec![Scale {
            aesthetic: "x".to_string(),
            scale_type: None,
            input_range: Some(vec![
                ArrayElement::String("A".to_string()),
                ArrayElement::String("B".to_string()),
            ]),
            explicit_input_range: false,
            output_range: None,
            transform: None,
            explicit_transform: false,
            properties: Parameters::new(),
            resolved: false,
            label_mapping: None,
            label_template: "{}".to_string(),
        }];
        let context = RenderContext::new(
            &scales,
            cartesian.as_ref(),
            AestheticContext::from_static(&["x", "y"], &[]),
        );
        let result = context.get_extent("x");
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("no valid numeric range"));
    }

    #[test]
    fn test_rule_renderer_multiple_diagonal_lines() {
        use crate::reader::{DuckDBReader, Reader};
        use crate::writer::{VegaLiteWriter, Writer};

        // Test that rule with 3 different slopes renders 3 separate lines
        let query = r#"
            WITH points AS (
                SELECT * FROM (VALUES (0, 5), (5, 15), (10, 25)) AS t(x, y)
            ),
            lines AS (
                SELECT * FROM (VALUES
                    (2, 5, 'A'),
                    (1, 10, 'B'),
                    (3, 0, 'C')
                ) AS t(slope, y, line_id)
            )
            SELECT * FROM points
            VISUALISE
            DRAW point MAPPING x AS x, y AS y
            DRAW rule MAPPING slope AS slope, y AS y, line_id AS color FROM lines
        "#;

        // Execute query
        let reader = DuckDBReader::from_connection_string("duckdb://memory")
            .expect("Failed to create reader");
        let spec = reader.execute(query).expect("Failed to execute query");

        // Render to Vega-Lite
        let writer = VegaLiteWriter::new();
        let vl_json = writer.render(&spec).expect("Failed to render spec");

        // Parse JSON
        let vl_spec: serde_json::Value =
            serde_json::from_str(&vl_json).expect("Failed to parse Vega-Lite JSON");

        // Verify we have 2 layers (point + rule)
        let layers = vl_spec["layer"].as_array().expect("No layers found");
        assert_eq!(layers.len(), 2, "Should have 2 layers (point + rule)");

        // Get the rule layer (second layer)
        let rule_layer = &layers[1];

        // Verify it's a rule mark
        assert_eq!(
            rule_layer["mark"]["type"], "rule",
            "Rule should use rule mark"
        );

        // Verify transforms exist
        let transforms = rule_layer["transform"]
            .as_array()
            .expect("No transforms found");

        // Should have 4 calculate transforms + 1 filter = 5 total
        assert_eq!(
            transforms.len(),
            5,
            "Should have 5 transforms (primary_min, primary_max, secondary_min, secondary_max, filter)"
        );

        // Verify primary_min/primary_max transforms exist with consistent naming
        let primary_min_transform = transforms
            .iter()
            .find(|t| t["as"] == "primary_min")
            .expect("primary_min transform not found");
        let primary_max_transform = transforms
            .iter()
            .find(|t| t["as"] == "primary_max")
            .expect("primary_max transform not found");

        assert!(
            primary_min_transform["calculate"].is_string(),
            "primary_min should have calculate expression"
        );
        assert!(
            primary_max_transform["calculate"].is_string(),
            "primary_max should have calculate expression"
        );

        // Verify secondary_min and secondary_max transforms use slope and intercept with primary_min/primary_max
        let secondary_min_transform = transforms
            .iter()
            .find(|t| t["as"] == "secondary_min")
            .expect("secondary_min transform not found");
        let secondary_max_transform = transforms
            .iter()
            .find(|t| t["as"] == "secondary_max")
            .expect("secondary_max transform not found");

        let secondary_min_calc = secondary_min_transform["calculate"]
            .as_str()
            .expect("secondary_min calculate should be string");
        let secondary_max_calc = secondary_max_transform["calculate"]
            .as_str()
            .expect("secondary_max calculate should be string");

        // Should reference slope, pos2 (acting as intercept for y-mapped rules), and primary_min/primary_max
        assert!(
            secondary_min_calc.contains("__ggsql_aes_pos2__"),
            "secondary_min should reference pos2 (y intercept)"
        );
        assert!(
            secondary_min_calc.contains("datum.primary_min"),
            "secondary_min should reference datum.primary_min"
        );
        assert!(
            secondary_max_calc.contains("__ggsql_aes_pos2__"),
            "secondary_max should reference pos2 (y intercept)"
        );
        assert!(
            secondary_max_calc.contains("datum.primary_max"),
            "secondary_max should reference datum.primary_max"
        );

        // Verify encoding has x, x2, y, y2 with consistent field names
        let encoding = rule_layer["encoding"]
            .as_object()
            .expect("No encoding found");

        assert!(encoding.contains_key("x"), "Should have x encoding");
        assert!(encoding.contains_key("x2"), "Should have x2 encoding");
        assert!(encoding.contains_key("y"), "Should have y encoding");
        assert!(encoding.contains_key("y2"), "Should have y2 encoding");

        // Verify consistent naming: primary_min/max for x, secondary_min/max for y (default orientation)
        assert_eq!(
            encoding["x"]["field"], "primary_min",
            "x should reference primary_min field"
        );
        assert_eq!(
            encoding["x2"]["field"], "primary_max",
            "x2 should reference primary_max field"
        );
        assert_eq!(
            encoding["y"]["field"], "secondary_min",
            "y should reference secondary_min field"
        );
        assert_eq!(
            encoding["y2"]["field"], "secondary_max",
            "y2 should reference secondary_max field"
        );

        // Verify stroke encoding exists for line_id (color aesthetic becomes stroke for rule mark)
        assert!(
            encoding.contains_key("stroke"),
            "Should have stroke encoding for line_id"
        );

        // Verify data has 3 rule rows (one per slope)
        let data_values = vl_spec["data"]["values"]
            .as_array()
            .expect("No data values found");

        let rule_rows: Vec<_> = data_values
            .iter()
            .filter(|row| {
                row["__ggsql_source__"] == "__ggsql_layer_1__"
                    && row["__ggsql_aes_slope__"].is_number()
            })
            .collect();

        assert_eq!(
            rule_rows.len(),
            3,
            "Should have 3 rule rows (3 different slopes)"
        );

        // Verify we have slopes 1, 2, 3
        let mut slopes: Vec<f64> = rule_rows
            .iter()
            .map(|row| row["__ggsql_aes_slope__"].as_f64().unwrap())
            .collect();
        slopes.sort_by(|a, b| a.partial_cmp(b).unwrap());

        assert_eq!(
            slopes,
            vec![1.0, 2.0, 3.0],
            "Should have slopes 1, 2, and 3"
        );
    }

    #[test]
    fn test_sloped_rule_renderer_horizontal_orientation() {
        use crate::reader::{DuckDBReader, Reader};
        use crate::writer::{VegaLiteWriter, Writer};

        // Test that sloped rule with x mapping (horizontal) infers y varies
        let query = r#"
            WITH points AS (
                SELECT * FROM (VALUES (0, 5), (5, 15), (10, 25)) AS t(x, y)
            ),
            lines AS (
                SELECT * FROM (VALUES (0.4, -1, 'A')) AS t(slope, x, line_id)
            )
            SELECT * FROM points
            VISUALISE
            DRAW point MAPPING x AS x, y AS y
            DRAW rule MAPPING slope AS slope, x AS x, line_id AS color FROM lines
        "#;

        // Execute query
        let reader = DuckDBReader::from_connection_string("duckdb://memory")
            .expect("Failed to create reader");
        let spec = reader.execute(query).expect("Failed to execute query");

        // Render to Vega-Lite
        let writer = VegaLiteWriter::new();
        let vl_json = writer.render(&spec).expect("Failed to render spec");

        // Parse JSON
        let vl_spec: serde_json::Value =
            serde_json::from_str(&vl_json).expect("Failed to parse Vega-Lite JSON");

        // Get the rule layer (second layer)
        let layers = vl_spec["layer"].as_array().expect("No layers found");
        let rule_layer = &layers[1];

        // Verify transforms exist
        let transforms = rule_layer["transform"]
            .as_array()
            .expect("No transforms found");

        // Verify primary_min/max use pos2 extent (y-axis) for horizontal (x-mapped) orientation
        let primary_min_transform = transforms
            .iter()
            .find(|t| t["as"] == "primary_min")
            .expect("primary_min transform not found");
        let primary_max_transform = transforms
            .iter()
            .find(|t| t["as"] == "primary_max")
            .expect("primary_max transform not found");

        // The primary extent should come from the y-axis for horizontal orientation
        assert!(
            primary_min_transform["calculate"].is_string(),
            "primary_min should have calculate expression"
        );
        assert!(
            primary_max_transform["calculate"].is_string(),
            "primary_max should have calculate expression"
        );

        // Verify secondary_min and secondary_max use pos1 (x intercept) for horizontal orientation
        let secondary_min_transform = transforms
            .iter()
            .find(|t| t["as"] == "secondary_min")
            .expect("secondary_min transform not found");
        let secondary_max_transform = transforms
            .iter()
            .find(|t| t["as"] == "secondary_max")
            .expect("secondary_max transform not found");

        let secondary_min_calc = secondary_min_transform["calculate"]
            .as_str()
            .expect("secondary_min calculate should be string");
        let secondary_max_calc = secondary_max_transform["calculate"]
            .as_str()
            .expect("secondary_max calculate should be string");

        // Should reference pos1 (x intercept) for horizontal orientation
        assert!(
            secondary_min_calc.contains("__ggsql_aes_pos1__"),
            "secondary_min should reference pos1 (x intercept)"
        );
        assert!(
            secondary_max_calc.contains("__ggsql_aes_pos1__"),
            "secondary_max should reference pos1 (x intercept)"
        );

        // Verify encoding has y as primary axis (mapped to primary_min/max)
        let encoding = rule_layer["encoding"]
            .as_object()
            .expect("No encoding found");

        // For horizontal orientation (x-mapped): y is primary (uses primary_min/max), x is secondary
        assert_eq!(
            encoding["y"]["field"], "primary_min",
            "y should reference primary_min field for horizontal orientation"
        );
        assert_eq!(
            encoding["y2"]["field"], "primary_max",
            "y2 should reference primary_max field for horizontal orientation"
        );
        assert_eq!(
            encoding["x"]["field"], "secondary_min",
            "x should reference secondary_min field for horizontal orientation"
        );
        assert_eq!(
            encoding["x2"]["field"], "secondary_max",
            "x2 should reference secondary_max field for horizontal orientation"
        );
    }

    #[test]
    fn test_path_renderer_varying_aesthetics_metadata() {
        use crate::df;
        use crate::plot::{AestheticValue, Geom, Layer};

        let renderer = PathRenderer;
        let mut layer = Layer::new(Geom::line());

        // Create DataFrame with varying stroke
        let pos1_col = naming::aesthetic_column("pos1");
        let pos2_col = naming::aesthetic_column("pos2");
        let df = df! {
            pos1_col.as_str() => vec![1.0, 2.0, 3.0],
            pos2_col.as_str() => vec![10.0, 20.0, 30.0],
            "color" => vec![1.0, 2.0, 3.0],
        }
        .unwrap();

        // Map stroke to color column (continuous, not in partition_by)
        layer.mappings.insert(
            "stroke".to_string(),
            AestheticValue::standard_column("color"),
        );

        // Prepare data - should detect varying stroke
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        match prepared {
            PreparedData::Single { metadata, .. } => {
                let varying_aesthetics = metadata
                    .downcast_ref::<Vec<&'static str>>()
                    .expect("Metadata should be Vec<&str>");
                assert_eq!(varying_aesthetics.len(), 1);
                assert!(varying_aesthetics.contains(&"stroke"));
            }
            _ => panic!("Expected Single variant"),
        }
    }

    #[test]
    fn test_path_renderer_trail_mark_for_varying_linewidth() {
        use crate::df;
        use crate::plot::{AestheticValue, Geom, Layer};

        let renderer = PathRenderer;
        let mut layer = Layer::new(Geom::line());

        // Create DataFrame with varying linewidth
        let pos1_col = naming::aesthetic_column("pos1");
        let pos2_col = naming::aesthetic_column("pos2");
        let linewidth_col = naming::aesthetic_column("linewidth");
        let df = df! {
            pos1_col.as_str() => vec![1.0, 2.0, 3.0],
            pos2_col.as_str() => vec![10.0, 20.0, 30.0],
            linewidth_col.as_str() => vec![1.0, 3.0, 5.0],
        }
        .unwrap();

        // Map linewidth to column
        layer.mappings.insert(
            "linewidth".to_string(),
            AestheticValue::standard_column(naming::aesthetic_column("linewidth")),
        );

        // Prepare data
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        // Create a mock layer spec
        let layer_spec = json!({
            "mark": {"type": "line", "clip": true},
            "encoding": {
                "x": {"field": naming::aesthetic_column("pos1"), "type": "quantitative"},
                "y": {"field": naming::aesthetic_column("pos2"), "type": "quantitative"},
                "strokeWidth": {"field": naming::aesthetic_column("linewidth"), "type": "quantitative"}
            }
        });

        // Finalize should switch to trail mark and translate encodings
        let context = RenderContext::default_for_test();
        let result = renderer
            .finalize(layer_spec.clone(), &layer, "test", &prepared, &context)
            .unwrap();

        assert_eq!(result.len(), 1);
        let spec = &result[0];

        // Check mark type is trail
        assert_eq!(spec["mark"]["type"], "trail");
        assert_eq!(spec["mark"]["strokeWidth"], 0);

        // Check encoding translations
        let encoding = spec["encoding"].as_object().unwrap();
        assert!(encoding.contains_key("size"), "Should have size encoding");
        assert!(
            !encoding.contains_key("strokeWidth"),
            "strokeWidth should be removed"
        );
        // No stroke mapping in this test, so no fill expected
        assert!(!encoding.contains_key("stroke"), "stroke should be removed");
    }

    #[test]
    fn test_path_renderer_trail_mark_with_stroke_legend() {
        use crate::df;
        use crate::plot::{AestheticValue, Geom, Layer};

        let context = RenderContext::default_for_test();
        let renderer = PathRenderer;
        let mut layer = Layer::new(Geom::line());

        // Create DataFrame with varying linewidth and stroke
        let pos1_col = naming::aesthetic_column("pos1");
        let pos2_col = naming::aesthetic_column("pos2");
        let linewidth_col = naming::aesthetic_column("linewidth");
        let stroke_col = naming::aesthetic_column("stroke");
        let df = df! {
            pos1_col.as_str() => vec![1.0, 2.0, 3.0],
            pos2_col.as_str() => vec![10.0, 20.0, 30.0],
            linewidth_col.as_str() => vec![1.0, 3.0, 5.0],
            stroke_col.as_str() => vec!["A", "A", "B"],
        }
        .unwrap();

        // Map linewidth and stroke to columns
        layer.mappings.insert(
            "linewidth".to_string(),
            AestheticValue::standard_column(naming::aesthetic_column("linewidth")),
        );
        layer.mappings.insert(
            "stroke".to_string(),
            AestheticValue::standard_column(naming::aesthetic_column("stroke")),
        );

        // Prepare data
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        // Create a mock layer spec with stroke legend
        let layer_spec = json!({
            "mark": {"type": "line", "clip": true},
            "encoding": {
                "x": {"field": naming::aesthetic_column("pos1"), "type": "quantitative"},
                "y": {"field": naming::aesthetic_column("pos2"), "type": "quantitative"},
                "strokeWidth": {"field": naming::aesthetic_column("linewidth"), "type": "quantitative"},
                "stroke": {
                    "field": naming::aesthetic_column("stroke"),
                    "type": "nominal",
                    "legend": {
                        "title": "direction"
                    }
                }
            }
        });

        // Finalize should switch to trail mark and translate encodings
        let result = renderer
            .finalize(layer_spec.clone(), &layer, "test", &prepared, &context)
            .unwrap();

        assert_eq!(result.len(), 1);
        let spec = &result[0];

        // Check mark type is trail
        assert_eq!(spec["mark"]["type"], "trail");
        assert_eq!(spec["mark"]["strokeWidth"], 0);

        // Check encoding translations
        let encoding = spec["encoding"].as_object().unwrap();
        assert!(encoding.contains_key("size"), "Should have size encoding");
        assert!(encoding.contains_key("fill"), "Should have fill encoding");
        assert!(!encoding.contains_key("stroke"), "stroke should be removed");

        // Check that fill legend has symbolStrokeColor
        let fill = &encoding["fill"];
        assert!(fill["legend"].is_object(), "fill should have legend");
        let legend = fill["legend"].as_object().unwrap();
        assert!(
            legend.contains_key("symbolStrokeColor"),
            "fill legend should have symbolStrokeColor"
        );
        assert_eq!(
            legend["symbolStrokeColor"]["expr"], "scale('fill', datum.value)",
            "symbolStrokeColor should use fill scale"
        );
    }

    #[test]
    fn test_path_renderer_segmentation_for_varying_stroke() {
        use crate::df;
        use crate::plot::{AestheticValue, Geom, Layer};

        let renderer = PathRenderer;
        let mut layer = Layer::new(Geom::line());

        // Create DataFrame with varying stroke
        let pos1_col = naming::aesthetic_column("pos1");
        let pos2_col = naming::aesthetic_column("pos2");
        let df = df! {
            pos1_col.as_str() => vec![1.0, 2.0, 3.0],
            pos2_col.as_str() => vec![10.0, 20.0, 30.0],
            "color" => vec![1.0, 2.0, 3.0],
            ROW_INDEX_COLUMN => vec![0i32, 1, 2],
        }
        .unwrap();

        // Map stroke to color column
        layer.mappings.insert(
            "stroke".to_string(),
            AestheticValue::standard_column("color"),
        );

        // Prepare data
        let prepared = renderer
            .prepare_data(&df, &layer, "test", &HashMap::new())
            .unwrap();

        // Create a mock layer spec
        let layer_spec = json!({
            "mark": {"type": "line", "clip": true},
            "encoding": {
                "x": {"field": naming::aesthetic_column("pos1"), "type": "quantitative"},
                "y": {"field": naming::aesthetic_column("pos2"), "type": "quantitative"},
                "stroke": {"field": "color", "type": "nominal"}
            }
        });

        // Finalize should apply segmentation transforms
        let context = RenderContext::default_for_test();
        let result = renderer
            .finalize(layer_spec.clone(), &layer, "test", &prepared, &context)
            .unwrap();

        assert_eq!(result.len(), 1);
        let spec = &result[0];

        // Check transforms exist
        let transforms = spec["transform"]
            .as_array()
            .expect("Should have transforms");
        assert!(!transforms.is_empty());

        // Check for window transform (lead operation)
        let has_window = transforms.iter().any(|t| t.get("window").is_some());
        assert!(has_window, "Should have window transform for lead");

        // Check for flatten transform
        let has_flatten = transforms.iter().any(|t| t.get("flatten").is_some());
        assert!(has_flatten, "Should have flatten transform");

        // Check for detail encoding with segment_id
        let encoding = spec["encoding"].as_object().unwrap();
        assert!(
            encoding.contains_key("detail"),
            "Should have detail encoding"
        );
        assert_eq!(
            encoding["detail"]["field"], "__segment_id__",
            "Detail should use segment_id"
        );

        // Check that x/y use _final fields
        assert!(
            encoding["x"]["field"].as_str().unwrap().ends_with("_final"),
            "x should use _final field"
        );
        assert!(
            encoding["y"]["field"].as_str().unwrap().ends_with("_final"),
            "y should use _final field"
        );
    }
}