ggsql 0.4.1

//! Vega-Lite JSON writer implementation
//!
//! Converts ggsql specifications and DataFrames into Vega-Lite JSON format
//! for web-based interactive visualizations.
//!
//! # Mapping Strategy
//!
//! - ggsql Geom -> Vega-Lite mark type
//! - ggsql aesthetics -> Vega-Lite encoding channels
//! - ggsql layers -> Vega-Lite layer composition
//! - Polars DataFrame -> Vega-Lite inline data
//!
//! # Example
//!
//! ```rust,ignore
//! use ggsql::writer::{Writer, VegaLiteWriter};
//!
//! let writer = VegaLiteWriter::new();
//! let vega_json = writer.write(&spec, &dataframe)?;
//! // Can be rendered in browser with vega-embed
//! ```

mod data;
mod encoding;
mod layer;
mod projection;

use crate::plot::ArrayElement;
use crate::plot::{ParameterValue, Parameters, Scale, ScaleTypeKind};
use crate::writer::Writer;
use crate::{naming, AestheticValue, DataFrame, GgsqlError, Plot, Result};
use serde_json::{json, Value};
use std::collections::HashMap;

// Re-export submodule functions for use in write()
use data::{collect_binned_columns, is_binned_aesthetic, unify_datasets};
use encoding::{
    build_detail_encoding, build_encoding_channel, infer_field_type, map_aesthetic_name,
};
use layer::{geom_to_mark, get_renderer, validate_layer_columns, GeomRenderer, PreparedData};
use projection::{get_projection_renderer, ProjectionRenderer};

/// Conversion factor from points to pixels (CSS standard: 96 DPI, 72 points/inch)
/// 1 point = 96/72 pixels = 1.333
const POINTS_TO_PIXELS: f64 = 96.0 / 72.0;

/// Conversion factor from radius (in points) to area (in square pixels)
/// Used for size aesthetic: area = pi * r^2 where r is in pixels
/// So: area_px^2 = pi * (r_pt * POINTS_TO_PIXELS)^2 = pi * r_pt^2 * (96/72)^2
const POINTS_TO_AREA: f64 = std::f64::consts::PI * POINTS_TO_PIXELS * POINTS_TO_PIXELS;

/// Default size (in pixels) for polar coordinate visualizations
/// Used as fallback when explicit dimensions are not specified
pub(super) const DEFAULT_POLAR_SIZE: f64 = 350.0;

/// Split a label string on newlines and return appropriate JSON value
///
/// Returns a JSON array if the string contains multiple lines, or a JSON string
/// if it's a single line. Handles both actual newlines (from database columns,
/// CHAR(10), imported data) and literal \\n (from SQL string literals).
fn split_label_on_newlines(label: &str) -> Value {
    // Normalize literal \\n to actual \n, then split on any newline type
    let normalized = label.replace("\\n", "\n");
    let lines: Vec<&str> = normalized.lines().collect();

    // Return array if multiple lines, string if single line
    if lines.len() > 1 {
        json!(lines)
    } else {
        json!(label)
    }
}

/// Result of preparing layer data for rendering
///
/// Contains the datasets, renderers, and prepared data needed to build Vega-Lite layers.
struct LayerPreparation {
    /// Individual datasets keyed by layer/component identifier
    datasets: serde_json::Map<String, Value>,
    /// Renderers for each layer (one per layer in spec.layers)
    renderers: Vec<Box<dyn GeomRenderer>>,
    /// Prepared data for each layer (one per layer in spec.layers)
    prepared: Vec<PreparedData>,
}

// =============================================================================
// Helper Functions
// =============================================================================

/// Prepare layer data using renderers
///
/// For each layer:
/// - Gets the appropriate renderer for the geom type
/// - Prepares data (handles both standard and composite cases like boxplot)
/// - Builds individual datasets map with type-specific keys
///
/// Returns the datasets map, renderers, and prepared data for layer building.
fn prepare_layer_data(
    spec: &Plot,
    data: &HashMap<String, DataFrame>,
    layer_data_keys: &[String],
    binned_columns: &HashMap<String, Vec<f64>>,
) -> Result<LayerPreparation> {
    let mut individual_datasets = serde_json::Map::new();
    let mut layer_renderers: Vec<Box<dyn GeomRenderer>> = Vec::new();
    let mut prepared_data: Vec<PreparedData> = Vec::new();

    for (layer_idx, layer) in spec.layers.iter().enumerate() {
        let data_key = &layer_data_keys[layer_idx];
        let df = data.get(data_key).ok_or_else(|| {
            GgsqlError::WriterError(format!(
                "Missing data source '{}' for layer {}",
                data_key,
                layer_idx + 1
            ))
        })?;

        // Get the appropriate renderer for this geom type
        let renderer = get_renderer(&layer.geom);

        // Prepare data using the renderer (handles both standard and composite cases)
        let prepared = renderer.prepare_data(df, layer, data_key, binned_columns)?;

        // Add data to individual datasets based on prepared type
        match &prepared {
            PreparedData::Single { values, .. } => {
                individual_datasets.insert(data_key.clone(), json!(values));
            }
            PreparedData::Composite { components, .. } => {
                // For composite geoms (boxplot, etc.), add each component dataset
                // with type-specific keys (e.g., "__ggsql_layer_0__lower_whisker")
                for (component_name, values) in components {
                    let type_key = format!("{}{}", data_key, component_name);
                    individual_datasets.insert(type_key, json!(values));
                }
            }
        }

        layer_renderers.push(renderer);
        prepared_data.push(prepared);
    }

    Ok(LayerPreparation {
        datasets: individual_datasets,
        renderers: layer_renderers,
        prepared: prepared_data,
    })
}

/// Build Vega-Lite layers from spec
///
/// For each layer:
/// - Creates layer spec with mark type
/// - Builds transform array with source filter
/// - Builds encoding channels for each aesthetic mapping
/// - Handles binned position aesthetics (x2/y2 channels)
/// - Adds aesthetic parameters from SETTING as literal encodings
/// - Adds detail encoding for partition_by columns
/// - Applies geom-specific modifications via renderer
/// - Finalizes layers (may expand composite geoms into multiple layers)
///
/// The `projection` determines how internal position aesthetics are mapped to
/// Vega-Lite encoding channel names.
fn build_layers(
    spec: &Plot,
    data: &HashMap<String, DataFrame>,
    layer_data_keys: &[String],
    layer_renderers: &[Box<dyn GeomRenderer>],
    prepared_data: &[PreparedData],
    projection: &dyn ProjectionRenderer,
) -> Result<Vec<Value>> {
    let mut layers = Vec::new();

    // Build context once for all layers
    let context =
        encoding::RenderContext::new(&spec.scales, projection, spec.get_aesthetic_context());

    for (layer_idx, layer) in spec.layers.iter().enumerate() {
        let data_key = &layer_data_keys[layer_idx];
        let df = data.get(data_key).unwrap();
        let renderer = &layer_renderers[layer_idx];
        let prepared = &prepared_data[layer_idx];

        // Layer spec with mark
        let mut layer_spec = json!({
            "mark": geom_to_mark(&layer.geom)
        });

        // Build transform array for this layer
        // Always starts with a filter to select this layer's data from unified dataset
        let mut transforms: Vec<Value> = Vec::new();

        // Add source filter transform (if the renderer needs it)
        // Composite geoms like boxplot add their own type-specific filters
        if renderer.needs_source_filter() {
            transforms.push(json!({
                "filter": {
                    "field": naming::SOURCE_COLUMN,
                    "equal": data_key
                }
            }));
        }

        // Set transform array on layer spec
        layer_spec["transform"] = json!(transforms);

        // Build encoding for this layer
        let mut encoding = build_layer_encoding(layer, df, spec, projection)?;

        // For point marks, remove fill: null from encoding — Vega-Lite point marks
        // are unfilled by default, so omitting it achieves the same visual result
        // without making legend symbols (e.g., size) invisible. Other mark types
        // (bar, area, etc.) are filled by default, so fill: null must be preserved.
        if layer.geom.geom_type() == crate::plot::layer::geom::GeomType::Point
            && encoding
                .get("fill")
                .and_then(|v| v.get("value"))
                .is_some_and(|v| v.is_null())
        {
            encoding.remove("fill");
        }

        layer_spec["encoding"] = Value::Object(encoding);

        // Apply geom-specific spec modifications via renderer
        renderer.modify_spec(&mut layer_spec, layer, &context)?;

        // Finalize the layer (may expand into multiple layers for composite geoms)
        let final_layers = renderer.finalize(layer_spec, layer, data_key, prepared, &context)?;
        layers.extend(final_layers);
    }

    Ok(layers)
}

/// Build encoding channels for a single layer
///
/// Handles:
/// - Tracking titled aesthetic families (one title per family)
/// - Building encoding channels for each aesthetic mapping
/// - Binned position aesthetics (x2/y2 channels for bin width)
/// - Aesthetic parameters from SETTING as literal encodings
/// - Detail encoding for partition_by columns
/// - Geom-specific encoding modifications via renderer
///
/// The `projection` determines how internal position aesthetics (pos1, pos2) are
/// mapped to Vega-Lite encoding channel names (x/y for cartesian, theta/radius for polar).
fn build_layer_encoding(
    layer: &crate::plot::Layer,
    df: &DataFrame,
    spec: &Plot,
    projection: &dyn ProjectionRenderer,
) -> Result<serde_json::Map<String, Value>> {
    let mut encoding = serde_json::Map::new();

    // Get aesthetic context for name transformation
    let aesthetic_ctx = spec.get_aesthetic_context();

    // Track which aesthetic families have been titled to ensure only one title per family
    let mut titled_families: std::collections::HashSet<String> = std::collections::HashSet::new();

    // Collect primary aesthetics that exist in the layer (for title handling)
    // e.g., if layer has "y", then "ymin" and "ymax" should suppress their titles
    let primary_aesthetics: std::collections::HashSet<String> = layer
        .mappings
        .aesthetics
        .keys()
        .filter(|a| {
            aesthetic_ctx
                .primary_internal_position(a)
                .map(|p| p == a.as_str())
                .unwrap_or(false)
        })
        .cloned()
        .collect();

    // Create encoding context for this layer
    let mut enc_ctx = encoding::EncodingContext {
        df,
        spec,
        titled_families: &mut titled_families,
        primary_aesthetics: &primary_aesthetics,
    };

    // Build encoding channels for each aesthetic mapping
    // Mappings contains:
    // 1. Column references from MAPPING clause (apply scales)
    // 2. Query literals from MAPPING (converted to columns, apply scales)
    // 3. Literals from SETTING/defaults (remain as literals, no scales)
    for (aesthetic, value) in &layer.mappings.aesthetics {
        // Skip facet aesthetics - they are handled via top-level facet structure,
        // not as encoding channels. Adding them to encoding would create row-based
        // faceting instead of the intended wrap/grid layout.
        // Check for internal facet names (facet1, facet2) since transformation has occurred.
        if aesthetic_ctx.is_internal_facet(aesthetic) {
            continue;
        }

        // Skip structural aesthetics that are consumed by renderers during data
        // preparation but are not visual encoding channels in Vega-Lite.
        if aesthetic == "geometry" || aesthetic == "type" {
            continue;
        }

        let mut channel_name = map_aesthetic_name(aesthetic, &aesthetic_ctx, projection);
        // Opacity is retargeted to the fill when fill is supported
        if channel_name == "opacity" && layer.mappings.contains_key("fill") {
            channel_name = "fillOpacity".to_string();
        }

        // Secondary position channels (x2, y2, theta2, radius2) and geographic
        // channels (longitude, latitude) only support field/datum/value in
        // Vega-Lite — not type, scale, axis, or title.
        if matches!(
            channel_name.as_str(),
            "x2" | "y2"
                | "theta2"
                | "radius2"
                | "longitude"
                | "latitude"
                | "longitude2"
                | "latitude2"
        ) {
            let secondary_encoding = match value {
                AestheticValue::Column { name: col, .. } => json!({"field": col}),
                AestheticValue::Literal(lit) => json!({"value": lit.to_json()}),
                AestheticValue::AnnotationColumn { name: col } => json!({"field": col}),
            };
            encoding.insert(channel_name, secondary_encoding);
            continue;
        }

        let channel_encoding = build_encoding_channel(aesthetic, value, &mut enc_ctx)?;
        encoding.insert(channel_name, channel_encoding);

        // For binned position aesthetics (pos1, pos2), add end channel with bin_end column
        // This enables proper bin width rendering in Vega-Lite (maps to x2/y2 channels)
        if aesthetic_ctx.is_primary_internal(aesthetic) && is_binned_aesthetic(aesthetic, spec) {
            if let AestheticValue::Column { name: col, .. } = value {
                let end_col = naming::bin_end_column(col);
                let end_aesthetic = format!("{}end", aesthetic); // "pos1end" or "pos2end"
                let end_channel = map_aesthetic_name(&end_aesthetic, &aesthetic_ctx, projection); // maps to "x2" or "y2" (or theta2/radius2 for polar)
                encoding.insert(end_channel, json!({"field": end_col}));
            }
        }
    }

    // Add detail encoding for partition_by columns (grouping)
    if let Some(detail) = build_detail_encoding(&layer.partition_by) {
        encoding.insert("detail".to_string(), detail);
    }

    // Build context for renderer (also provides resolved position channel names)
    let context =
        encoding::RenderContext::new(&spec.scales, projection, spec.get_aesthetic_context());
    let (_, _, pos1_offset, pos2, _, pos2_offset) = &context.channels;

    // Add pos1 offset encoding for dodged positions (pos1offset column)
    // This column is created by position::apply_dodge() for Position::Dodge
    // The offset values are centered around 0 (e.g., -0.3, 0, +0.3 for 3 groups)
    // We set domain [-0.5, 0.5] to ensure the scale is symmetric and maps to full band width
    let pos1offset_col = naming::aesthetic_column("pos1offset");
    if df.column(&pos1offset_col).is_ok() {
        encoding.insert(
            pos1_offset.clone(),
            json!({
                "field": pos1offset_col,
                "type": "quantitative",
                "scale": {
                    "domain": [-0.5, 0.5]
                }
            }),
        );
    }

    // Add pos2 offset encoding for vertical jitter (pos2offset column)
    // This column is created by position::Jitter when pos2 axis is discrete
    let pos2offset_col = naming::aesthetic_column("pos2offset");
    if df.column(&pos2offset_col).is_ok() {
        encoding.insert(
            pos2_offset.clone(),
            json!({
                "field": pos2offset_col,
                "type": "quantitative",
                "scale": {
                    "domain": [-0.5, 0.5]
                }
            }),
        );
    }

    // Disable Vega-Lite's automatic stacking - we handle position adjustments ourselves.
    // This prevents Vega-Lite from applying its own stack/dodge logic on top of ours.
    // Only set stack: null on primary position channels (y/radius) — Vega-Lite does
    // not support 'stack' on secondary channels (y2/radius2) and Altair rejects it.
    // Geographic channels (latitude) don't support stack either.
    if !matches!(pos2.as_str(), "latitude") {
        if let Some(y_enc) = encoding.get_mut(pos2.as_str()) {
            if let Some(obj) = y_enc.as_object_mut() {
                obj.insert("stack".to_string(), Value::Null);
            }
        }
    }

    // Apply geom-specific encoding modifications via renderer
    let renderer = get_renderer(&layer.geom);
    renderer.modify_encoding(&mut encoding, layer, &context)?;

    Ok(encoding)
}

/// Apply faceting to Vega-Lite spec
///
/// Handles:
/// - FACET vars (wrap layout)
/// - FACET rows BY columns (grid layout)
/// - Moves layers into nested `spec` object
/// - Uses aesthetic column names (e.g., __ggsql_aes_facet1__)
/// - Respects scale types (Binned facets use bin: "binned")
/// - Scale resolution (scales property)
/// - Label renaming (RENAMING clause)
/// - Additional properties (ncol, etc.)
fn apply_faceting(
    vl_spec: &mut Value,
    facet: &crate::plot::Facet,
    facet_df: &DataFrame,
    scales: &[Scale],
    projection: &dyn ProjectionRenderer,
) {
    use crate::plot::FacetLayout;

    match &facet.layout {
        FacetLayout::Wrap { variables: _ } => {
            let facet_def = apply_facet_aesthetic(vl_spec, facet_df, scales, "facet1");
            vl_spec["facet"] = facet_def;
        }
        FacetLayout::Grid { row: _, column: _ } => {
            let mut facet_spec = serde_json::Map::new();

            let row_aes_col = naming::aesthetic_column("facet1");
            if facet_df.column(&row_aes_col).is_ok() {
                let row_def = apply_facet_aesthetic(vl_spec, facet_df, scales, "facet1");
                facet_spec.insert("row".to_string(), row_def);
            }

            // Column facet: use internal aesthetic column "facet2"
            // Vega-Lite requires "column" key in the facet object
            let col_aes_col = naming::aesthetic_column("facet2");
            if facet_df.column(&col_aes_col).is_ok() {
                let col_def = apply_facet_aesthetic(vl_spec, facet_df, scales, "facet2");
                facet_spec.insert("column".to_string(), col_def);
            }

            vl_spec["facet"] = Value::Object(facet_spec);
        }
    }

    // Move layer + width/height into inner spec (FacetSpec disallows
    // top-level layer/width/height in Vega-Lite v6)
    let mut spec_inner = json!({});
    if let Some(layer) = vl_spec.get("layer") {
        spec_inner["layer"] = layer.clone();
    }
    if let Some(w) = vl_spec.get("width").cloned() {
        spec_inner["width"] = w;
    }
    if let Some(h) = vl_spec.get("height").cloned() {
        spec_inner["height"] = h;
    }

    vl_spec["spec"] = spec_inner;
    let obj = vl_spec.as_object_mut().unwrap();
    obj.remove("layer");
    obj.remove("width");
    obj.remove("height");

    apply_facet_scale_resolution(vl_spec, &facet.properties, projection);
    apply_facet_properties(vl_spec, &facet.properties, facet.is_wrap());
}

/// Build a fully resolved facet field definition for a single facet aesthetic.
///
/// Combines all per-field steps: build the base definition, resolve and apply
/// panel ordering (rewriting data values to integers), apply label renaming
/// via header.labelExpr, and handle binned reverse sort.
fn apply_facet_aesthetic(
    vl_spec: &mut Value,
    facet_df: &DataFrame,
    scales: &[Scale],
    aesthetic: &str,
) -> Value {
    let aes_col = naming::aesthetic_column(aesthetic);
    let scale = scales.iter().find(|s| s.aesthetic == aesthetic);

    let mut facet_def = build_facet_field_def(facet_df, &aes_col, scale);

    let index_map = resolve_facet_ordering(scale);
    apply_facet_ordering(vl_spec, &aes_col, &index_map);

    let label_mapping = scale.and_then(|s| s.label_mapping.as_ref());
    apply_facet_label_renaming(&mut facet_def, label_mapping, scale, &index_map);

    apply_binned_facet_reverse(&mut facet_def, scale);

    facet_def
}

/// Build a facet field definition with proper type.
///
/// Facets always use "type": "nominal" since facet values are categorical
/// (even for binned data, the bin labels are discrete categories).
fn build_facet_field_def(df: &DataFrame, col: &str, scale: Option<&Scale>) -> Value {
    let mut field_def = json!({
        "field": col,
    });

    if let Some(scale) = scale {
        if let Some(ref scale_type) = scale.scale_type {
            match scale_type.scale_type_kind() {
                // All scale types use nominal for facets - the data column contains
                // categorical values (bin midpoints for binned, categories for discrete)
                ScaleTypeKind::Binned
                | ScaleTypeKind::Discrete
                | ScaleTypeKind::Ordinal
                | ScaleTypeKind::Continuous
                | ScaleTypeKind::Identity => {
                    field_def["type"] = json!("nominal");
                    return field_def;
                }
            }
        }
    }

    // Fall back to column type inference
    field_def["type"] = json!(infer_field_type(df, col));
    field_def
}

/// Resolve facet ordering into a value → index mapping.
///
/// For discrete facets: uses input_range (FROM clause) or breaks array order.
/// For binned facets: returns empty (bin midpoints already sort correctly).
///
/// The returned mapping is consumed by `apply_facet_ordering` (to rewrite the
/// data column to integers) and `apply_facet_label_renaming` (to build a
/// `header.labelExpr` that restores the display labels).
fn resolve_facet_ordering(scale: Option<&Scale>) -> Vec<(Value, usize)> {
    let Some(scale) = scale else {
        return Vec::new();
    };

    let is_reversed = matches!(
        scale.properties.get("reverse"),
        Some(ParameterValue::Boolean(true))
    );

    let is_binned = scale
        .scale_type
        .as_ref()
        .map(|st| st.scale_type_kind() == ScaleTypeKind::Binned)
        .unwrap_or(false);

    if is_binned {
        // Binned facets use numeric midpoints that sort naturally — no index
        // rewriting needed. Reversed binned facets are handled separately via
        // apply_binned_facet_reverse on the facet definition.
        return Vec::new();
    }

    // For discrete facets: use input_range (FROM clause) if present,
    // otherwise fall back to breaks property
    let order_values: Vec<ArrayElement> = if let Some(ref input_range) = scale.input_range {
        // Use explicit input_range from FROM clause
        input_range.clone()
    } else if let Some(ParameterValue::Array(arr)) = scale.properties.get("breaks") {
        // Fall back to breaks if present
        arr.clone()
    } else {
        return Vec::new();
    };

    // Convert to JSON values, preserving null
    let mut sort_values: Vec<Value> = order_values.iter().map(|e| e.to_json()).collect();

    // Apply reverse if specified
    if is_reversed {
        sort_values.reverse();
    }

    sort_values
        .iter()
        .enumerate()
        .map(|(i, v)| (v.clone(), i))
        .collect()
}

/// For binned facets with `reverse => true`, set `sort: "descending"` on the
/// facet definition. Discrete facets handle reversal through `resolve_facet_ordering`
/// (which reverses the index map), so this only applies to binned facets.
fn apply_binned_facet_reverse(facet_def: &mut Value, scale: Option<&Scale>) {
    let Some(scale) = scale else { return };

    let is_binned = scale
        .scale_type
        .as_ref()
        .map(|st| st.scale_type_kind() == ScaleTypeKind::Binned)
        .unwrap_or(false);

    if !is_binned {
        return;
    }

    if matches!(
        scale.properties.get("reverse"),
        Some(ParameterValue::Boolean(true))
    ) {
        facet_def["sort"] = json!("descending");
    }
}

/// Rewrite facet column values in `vl_spec.data.values` to integer sort indices.
///
/// Uses the mapping produced by `resolve_facet_ordering`. Values not present in
/// the mapping are assigned an index past the end (so they sort last).
///
/// Using integers instead of a literal `sort` array works around a Vega-Lite bug
/// where explicit sort arrays on facet definitions cause data-panel misalignment
/// (https://github.com/vega/vega-lite/issues/8675).
fn apply_facet_ordering(vl_spec: &mut Value, facet_col: &str, index_map: &[(Value, usize)]) {
    if index_map.is_empty() {
        return;
    }

    if let Some(data) = vl_spec
        .get_mut("data")
        .and_then(|d| d.get_mut("values"))
        .and_then(|v| v.as_array_mut())
    {
        let fallback = index_map.len();
        for row in data.iter_mut() {
            if let Some(obj) = row.as_object_mut() {
                let orig = obj.get(facet_col).cloned().unwrap_or(Value::Null);
                let idx = index_map
                    .iter()
                    .find(|(v, _)| *v == orig)
                    .map(|(_, i)| *i)
                    .unwrap_or(fallback);
                obj.insert(facet_col.to_string(), json!(idx));
            }
        }
    }
}

/// Apply scale resolution to Vega-Lite spec based on facet free property
///
/// Maps ggsql free property (boolean array) to Vega-Lite resolve.scale configuration:
/// - `[false, false]`: shared scales (Vega-Lite default, no resolve needed)
/// - `[true, false]`: independent pos1 scale (x or radius), shared pos2 scale
/// - `[false, true]`: shared pos1 scale, independent pos2 scale (y or theta)
/// - `[true, true]`: independent scales for both axes
///
/// The channel names depend on the projection:
/// - Cartesian: pos1 -> "x", pos2 -> "y"
/// - Polar: pos1 -> "radius", pos2 -> "theta"
fn apply_facet_scale_resolution(
    vl_spec: &mut Value,
    properties: &Parameters,
    projection: &dyn ProjectionRenderer,
) {
    let Some(ParameterValue::Array(arr)) = properties.get("free") else {
        // No free property means fixed/shared scales (Vega-Lite default)
        return;
    };

    // Determine channel names from the projection renderer
    let (pos1_channel, pos2_channel) = projection.position_channels();

    // Extract booleans from the array (position-indexed)
    let free_pos1 = arr
        .first()
        .map(|e| matches!(e, crate::plot::ArrayElement::Boolean(true)))
        .unwrap_or(false);
    let free_pos2 = arr
        .get(1)
        .map(|e| matches!(e, crate::plot::ArrayElement::Boolean(true)))
        .unwrap_or(false);

    // Apply resolve configuration to Vega-Lite spec
    if free_pos1 && free_pos2 {
        vl_spec["resolve"] = json!({
            "scale": {pos1_channel: "independent", pos2_channel: "independent"}
        });
    } else if free_pos1 {
        vl_spec["resolve"] = json!({
            "scale": {pos1_channel: "independent"}
        });
    } else if free_pos2 {
        vl_spec["resolve"] = json!({
            "scale": {pos2_channel: "independent"}
        });
    }
    // If neither is free, don't add resolve (Vega-Lite default is shared)
}

/// Apply label renaming to a facet definition via header.labelExpr
///
/// Uses Vega expression to transform facet labels:
/// - For discrete facets (non-empty `index_map`): the facet column has been
///   converted to integer sort indices by `apply_facet_ordering`. The labelExpr
///   maps those integers back to display labels, applying user renaming
///   (RENAMING clause) when present.
/// - For binned facets: uses build_symbol_legend_label_mapping for range-style labels
/// - NULL values suppress labels (maps to empty string)
///
/// Note: Wildcard templates are resolved during facet property resolution,
/// so by this point label_mapping contains all expanded mappings.
fn apply_facet_label_renaming(
    facet_def: &mut Value,
    label_mapping: Option<&HashMap<String, Option<String>>>,
    scale: Option<&Scale>,
    index_map: &[(Value, usize)],
) {
    if !index_map.is_empty() {
        // Facet column was converted to integers — build labelExpr mapping
        // index → display label (applying user renaming when present).
        let label_expr = build_indexed_facet_label_expr(index_map, label_mapping);
        facet_def["header"] = json!({ "labelExpr": label_expr });
        return;
    }

    // No index map: original (non-ordered or binned) path.
    // Only apply if there's a label mapping
    let has_mapping = label_mapping.is_some_and(|m| !m.is_empty());
    if !has_mapping {
        return;
    }

    // Check if this is a binned facet
    let is_binned = scale
        .and_then(|s| s.scale_type.as_ref())
        .map(|st| st.scale_type_kind() == ScaleTypeKind::Binned)
        .unwrap_or(false);

    let label_expr = if is_binned {
        // For binned facets: reuse build_symbol_legend_label_mapping and build_label_expr
        build_binned_facet_label_expr(label_mapping, scale)
    } else {
        // For discrete facets: compare datum.value against string values
        build_discrete_facet_label_expr(label_mapping)
    };

    // Add to facet definition
    facet_def["header"] = json!({ "labelExpr": label_expr });
}

/// Build labelExpr for binned facet values.
///
/// For binned facets, `datum.value` contains the bin midpoint (e.g., 25 for bin [20-30)).
/// This function maps midpoint values to range-style labels like "Lower – Upper",
/// using custom labels from label_mapping when available.
///
/// Unlike `build_symbol_legend_label_mapping` which maps Vega-Lite's auto-generated
/// range labels, this function maps numeric midpoints to our range labels.
fn build_binned_facet_label_expr(
    label_mapping: Option<&HashMap<String, Option<String>>>,
    scale: Option<&Scale>,
) -> String {
    let Some(scale) = scale else {
        return "datum.value".to_string();
    };

    let breaks = match scale.properties.get("breaks") {
        Some(ParameterValue::Array(arr)) => arr,
        _ => return "datum.value".to_string(),
    };

    if breaks.len() < 2 {
        return "datum.value".to_string();
    }

    // Get closed property for determining open-format labels
    let closed = scale
        .properties
        .get("closed")
        .and_then(|v| match v {
            ParameterValue::String(s) => Some(s.as_str()),
            _ => None,
        })
        .unwrap_or("left");

    let num_bins = breaks.len() - 1;

    // Build mapping from midpoint to range label
    let mut midpoint_to_range: Vec<(String, Option<String>)> = Vec::new();

    for i in 0..num_bins {
        let lower = &breaks[i];
        let upper = &breaks[i + 1];

        // Calculate midpoint for comparison
        let midpoint_str = calculate_midpoint_string(lower, upper, scale.transform.as_ref());
        let Some(midpoint_str) = midpoint_str else {
            continue;
        };

        // Get break values as strings (for default labels)
        let lower_str = lower.to_key_string();
        let upper_str = upper.to_key_string();

        // Build the range label
        let range_label = if let Some(label_mapping) = label_mapping {
            // Check if terminals are suppressed
            let lower_suppressed = label_mapping.get(&lower_str) == Some(&None);
            let upper_suppressed = label_mapping.get(&upper_str) == Some(&None);

            // Get custom labels (fall back to break values)
            let lower_label = label_mapping
                .get(&lower_str)
                .cloned()
                .flatten()
                .unwrap_or_else(|| lower_str.clone());
            let upper_label = label_mapping
                .get(&upper_str)
                .cloned()
                .flatten()
                .unwrap_or_else(|| upper_str.clone());

            // Determine label format based on terminal suppression
            if i == 0 && lower_suppressed {
                // First bin with suppressed lower terminal → open format
                let symbol = if closed == "right" { "≤" } else { "<" };
                Some(format!("{} {}", symbol, upper_label))
            } else if i == num_bins - 1 && upper_suppressed {
                // Last bin with suppressed upper terminal → open format
                let symbol = if closed == "right" { ">" } else { "≥" };
                Some(format!("{} {}", symbol, lower_label))
            } else {
                // Standard range format: "lower – upper"
                Some(format!("{} – {}", lower_label, upper_label))
            }
        } else {
            // No label mapping - use default range format with break values
            Some(format!("{} – {}", lower_str, upper_str))
        };

        midpoint_to_range.push((midpoint_str, range_label));
    }

    if midpoint_to_range.is_empty() {
        return "datum.value".to_string();
    }

    // Build labelExpr comparing datum.value against midpoints
    build_binned_facet_value_expr(&midpoint_to_range)
}

/// Build labelExpr comparing datum.value against midpoint values
fn build_binned_facet_value_expr(mappings: &[(String, Option<String>)]) -> String {
    let mut expr_parts: Vec<String> = Vec::new();

    for (midpoint, label) in mappings {
        // Compare as number for numeric midpoints, string for temporal
        let condition = format!("datum.value == {}", midpoint);
        let result = match label {
            Some(l) => format!("'{}'", escape_vega_string(l)),
            None => "''".to_string(),
        };
        expr_parts.push(format!("{} ? {}", condition, result));
    }

    if expr_parts.is_empty() {
        return "datum.value".to_string();
    }

    // Chain: cond1 ? val1 : cond2 ? val2 : datum.value
    let mut expr = "datum.value".to_string();
    for part in expr_parts.into_iter().rev() {
        expr = format!("{} : {}", part, expr);
    }
    expr
}

/// Calculate the midpoint string for a bin
fn calculate_midpoint_string(
    lower: &ArrayElement,
    upper: &ArrayElement,
    transform: Option<&crate::plot::scale::Transform>,
) -> Option<String> {
    match (lower, upper) {
        (ArrayElement::Number(l), ArrayElement::Number(u)) => {
            let midpoint = (*l + *u) / 2.0;

            // Check if temporal transform - format as ISO string (quoted for comparison)
            if let Some(t) = transform {
                if let Some(iso) = t.format_as_iso(midpoint) {
                    return Some(format!("'{}'", iso));
                }
            }

            // Numeric: format without trailing decimals if whole number
            Some(if midpoint.fract() == 0.0 {
                format!("{}", midpoint as i64)
            } else {
                format!("{}", midpoint)
            })
        }
        // Temporal ArrayElements - calculate midpoint and format as ISO (quoted)
        (ArrayElement::Date(l), ArrayElement::Date(u)) => {
            let midpoint = ((*l as f64) + (*u as f64)) / 2.0;
            Some(format!("'{}'", ArrayElement::date_to_iso(midpoint as i32)))
        }
        (ArrayElement::DateTime(l), ArrayElement::DateTime(u)) => {
            let midpoint = ((*l as f64) + (*u as f64)) / 2.0;
            Some(format!(
                "'{}'",
                ArrayElement::datetime_to_iso(midpoint as i64)
            ))
        }
        (ArrayElement::Time(l), ArrayElement::Time(u)) => {
            let midpoint = ((*l as f64) + (*u as f64)) / 2.0;
            Some(format!("'{}'", ArrayElement::time_to_iso(midpoint as i64)))
        }
        _ => None,
    }
}

/// Build labelExpr for integer-indexed facet values.
///
/// The `index_map` contains (original_value, sort_index) pairs produced by
/// `apply_facet_ordering`. Each entry becomes a ternary branch mapping the
/// integer back to a display label. When `label_mapping` provides a rename
/// for an original value, the renamed label is used.
fn build_indexed_facet_label_expr(
    index_map: &[(Value, usize)],
    label_mapping: Option<&HashMap<String, Option<String>>>,
) -> String {
    let mut expr_parts: Vec<String> = Vec::new();

    for (orig_value, idx) in index_map {
        let condition = format!("datum.value == {}", idx);

        // Determine the display label: check user renaming first, else use original
        let key = match orig_value {
            Value::Null => "null".to_string(),
            Value::String(s) => s.clone(),
            other => other.to_string(),
        };

        let label = label_mapping
            .and_then(|m| m.get(&key))
            .cloned()
            .unwrap_or(Some(key));

        let result = match label {
            Some(l) => format!("'{}'", escape_vega_string(&l)),
            None => "''".to_string(),
        };
        expr_parts.push(format!("{} ? {}", condition, result));
    }

    let mut expr = "datum.value".to_string();
    for part in expr_parts.into_iter().rev() {
        expr = format!("{} : {}", part, expr);
    }
    expr
}

/// Build labelExpr for discrete facet values.
fn build_discrete_facet_label_expr(
    label_mapping: Option<&HashMap<String, Option<String>>>,
) -> String {
    let Some(mappings) = label_mapping else {
        return "datum.value".to_string();
    };

    // Build labelExpr for Vega-Lite
    let mut expr_parts: Vec<String> = Vec::new();

    // Add explicit mappings
    for (from, to) in mappings {
        // Handle null values: 'null' key maps to JSON null comparison
        let condition = if from == "null" {
            "datum.value == null".to_string()
        } else {
            format!("datum.value == '{}'", escape_vega_string(from))
        };
        let result = match to {
            Some(label) => format!("'{}'", escape_vega_string(label)),
            None => "''".to_string(), // NULL suppresses label
        };
        expr_parts.push(format!("{} ? {}", condition, result));
    }

    // Default case: show original value
    let default_expr = "datum.value".to_string();

    // Build the full expression as nested ternary
    if expr_parts.is_empty() {
        default_expr
    } else {
        // Chain conditions: cond1 ? val1 : cond2 ? val2 : default
        let mut expr = default_expr;
        for part in expr_parts.into_iter().rev() {
            expr = format!("{} : {}", part, expr);
        }
        expr
    }
}

/// Escape a string for use in Vega expressions
pub(super) fn escape_vega_string(s: &str) -> String {
    s.replace('\\', "\\\\").replace('\'', "\\'")
}

/// Apply additional facet properties to Vega-Lite spec
///
/// Handles:
/// - ncol: Number of columns for wrap facets (maps to Vega-Lite's "columns")
///
/// Note: free is handled separately by apply_facet_scale_resolution
fn apply_facet_properties(vl_spec: &mut Value, properties: &Parameters, is_wrap: bool) {
    for (name, value) in properties {
        match name.as_str() {
            "ncol" if is_wrap => {
                // ncol maps to Vega-Lite's "columns" property
                if let ParameterValue::Number(n) = value {
                    vl_spec["columns"] = json!(*n as i64);
                }
            }
            "free" => {
                // Handled by apply_facet_scale_resolution
            }
            _ => {
                // Unknown properties ignored (resolution should have validated)
            }
        }
    }
}

/// Vega-Lite schema URL. Update this, the vendored schema file, and
/// the `include_str!` path in `VL_SCHEMA` when bumping versions.
const VEGALITE_SCHEMA: &str = "https://vega.github.io/schema/vega-lite/v6.json";

/// Vega-Lite JSON writer
///
/// Generates Vega-Lite v6 specifications from ggsql specs and data.
pub struct VegaLiteWriter {
    schema: String,
}

impl VegaLiteWriter {
    pub fn new() -> Self {
        Self {
            schema: VEGALITE_SCHEMA.to_string(),
        }
    }

    /// Build default Vega-Lite config matching ggplot2's theme_gray()
    ///
    /// Font sizes converted from ggplot2 points to pixels (1 pt ≈ 1.33 px at 96 DPI):
    /// - axis.text: 8.8 pts (rel(0.8) × 11) → 12 px
    /// - axis.title: 11 pts → 15 px
    /// - legend.text: 8.8 pts → 12 px
    /// - legend.title: 11 pts → 15 px
    /// - plot.title: 13.2 pts (rel(1.2) × 11) → 18 px
    /// - tick size: ~2.75 pts → 4 px
    fn default_theme_config(&self) -> Value {
        json!({
            "view": {
                "stroke": null,
                "fill": "#EBEBEB"
            },
            "axis": {
                "domain": false,
                "grid": true,
                "gridColor": "#FFFFFF",
                "gridWidth": 1,
                "tickColor": "#333333",
                "tickSize": 4,
                "labelColor": "#4D4D4D",
                "labelFontSize": 12,
                "titleColor": "#000000",
                "titleFontSize": 15,
                "titleFontWeight": "normal",
                "titlePadding": 10
            },
            "legend": {
                "labelColor": "#4D4D4D",
                "labelFontSize": 12,
                "titleColor": "#000000",
                "titleFontSize": 15,
                "titleFontWeight": "normal",
                "titlePadding": 8,
                "rowPadding": 6
            },
            "title": {
                "color": "#000000",
                "fontSize": 18,
                "fontWeight": "normal",
                "subtitleColor": "#4D4D4D",
                "subtitleFontSize": 15,
                "subtitleFontWeight": "normal",
                "anchor": "start",
                "frame": "group",
                "offset": 10
            },
            "header": {
                "labelColor": "#000000",
                "labelFontSize": 15,
                "labelFontWeight": "normal",
                "labelPadding": 5,
                "title": null
            }
        })
    }
}

impl Default for VegaLiteWriter {
    fn default() -> Self {
        Self::new()
    }
}

impl Writer for VegaLiteWriter {
    type Output = String;

    fn write(&self, spec: &Plot, data: &HashMap<String, DataFrame>) -> Result<String> {
        // 1. Validate spec
        self.validate(spec)?;

        // 2. Determine layer data keys
        let layer_data_keys: Vec<String> = spec
            .layers
            .iter()
            .enumerate()
            .map(|(idx, layer)| {
                layer
                    .data_key
                    .clone()
                    .unwrap_or_else(|| naming::layer_key(idx))
            })
            .collect();

        // 4. Validate columns for each layer
        let aesthetic_ctx = spec.get_aesthetic_context();
        for (layer_idx, (layer, key)) in spec.layers.iter().zip(layer_data_keys.iter()).enumerate()
        {
            let df = data.get(key).ok_or_else(|| {
                GgsqlError::WriterError(format!(
                    "Missing data source '{}' for layer {}",
                    key,
                    layer_idx + 1
                ))
            })?;
            validate_layer_columns(layer, df, layer_idx, &aesthetic_ctx)?;
        }

        // 5. Build base Vega-Lite spec
        let mut vl_spec = json!({
            "$schema": self.schema
        });
        // Get projection renderer (single instance used throughout)
        let projection =
            get_projection_renderer(spec.project.as_ref(), spec.facet.as_ref(), &spec.scales);

        if let Some((w, h)) = projection.panel_size() {
            vl_spec["width"] = w;
            vl_spec["height"] = h;
        }

        if let Some(labels) = &spec.labels {
            let title = labels.labels.get("title").and_then(|v| v.as_ref());
            let subtitle = labels.labels.get("subtitle").and_then(|v| v.as_ref());
            match (title, subtitle) {
                (Some(t), Some(st)) => {
                    // Vega-Lite uses an object for title + subtitle
                    // Split both title and subtitle on newlines
                    vl_spec["title"] = json!({
                        "text": split_label_on_newlines(t),
                        "subtitle": split_label_on_newlines(st)
                    });
                }
                (Some(t), None) => {
                    vl_spec["title"] = split_label_on_newlines(t);
                }
                (None, Some(st)) => {
                    vl_spec["title"] = json!({
                        "text": "",
                        "subtitle": split_label_on_newlines(st)
                    });
                }
                (None, None) => {}
            }
        }

        // 6. Collect binned columns
        let binned_columns = collect_binned_columns(spec);

        // 7. Prepare layer data
        let prep = prepare_layer_data(spec, data, &layer_data_keys, &binned_columns)?;

        // 8. Unify datasets
        let unified_data = unify_datasets(&prep.datasets)?;
        vl_spec["data"] = json!({"values": unified_data});

        // 9. Build layers
        let layers = build_layers(
            spec,
            data,
            &layer_data_keys,
            &prep.renderers,
            &prep.prepared,
            projection.as_ref(),
        )?;
        vl_spec["layer"] = json!(layers);

        // 10. Apply faceting
        if let Some(facet) = &spec.facet {
            let facet_df = data.get(&layer_data_keys[0]).unwrap();
            apply_faceting(
                &mut vl_spec,
                facet,
                facet_df,
                &spec.scales,
                projection.as_ref(),
            );
        }

        // 11. Apply projection (transforms + panel decoration)
        let mut theme = self.default_theme_config();
        projection.apply_projection(spec, &mut theme, &mut vl_spec)?;
        vl_spec["config"] = theme;

        // 12. Serialize
        serde_json::to_string_pretty(&vl_spec).map_err(|e| {
            GgsqlError::WriterError(format!("Failed to serialize Vega-Lite JSON: {}", e))
        })
    }

    fn validate(&self, spec: &Plot) -> Result<()> {
        // Check that we have at least one layer
        if spec.layers.is_empty() {
            return Err(GgsqlError::ValidationError(
                "VegaLiteWriter requires at least one layer".to_string(),
            ));
        }

        // Validate each layer
        for layer in &spec.layers {
            // Check aesthetics
            // We already checked the supported aesthetics during execution.
            // They'd fail now because delayed have to
            layer
                .validate_mapping(&spec.aesthetic_context, true)
                .map_err(|e| {
                    GgsqlError::ValidationError(format!("Layer validation failed: {}", e))
                })?;
            // Note: validate_settings() is already called during execution, before
            // internal parameters like "orientation" are set, so we don't call it here.
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::df;
    use crate::plot::{Labels, Layer, ParameterValue};
    use crate::Geom;
    use serde_json::Value;
    use std::collections::HashMap;
    use std::sync::LazyLock;

    // Re-export test functions from submodules
    use super::data::find_bin_for_value;
    use super::encoding::infer_field_type;
    use super::layer::geom_to_mark;

    fn sanitize_def_name(s: &str) -> String {
        s.chars()
            .map(|c| match c {
                '<' | '>' | '(' | ')' | '|' | '[' | ']' | '"' | ' ' | '{' | '}' => '_',
                _ => c,
            })
            .collect()
    }

    fn rewrite_refs(val: &mut Value) {
        match val {
            Value::Object(obj) => {
                if let Some(Value::String(s)) = obj.get_mut("$ref") {
                    if let Some(defname) = s.strip_prefix("#/definitions/") {
                        let sanitized = sanitize_def_name(defname);
                        if sanitized != defname {
                            *s = format!("#/definitions/{}", sanitized);
                        }
                    }
                }
                for v in obj.values_mut() {
                    rewrite_refs(v);
                }
            }
            Value::Array(arr) => arr.iter_mut().for_each(rewrite_refs),
            _ => {}
        }
    }

    /// The VL v6 schema uses TypeScript-style generics in definition names (e.g.,
    /// `MarkPropDef<(Gradient|string|null)>`) which are invalid in `$ref` URIs.
    /// Rename those definitions and rewrite their `$ref` references.
    fn sanitize_vegalite_schema(schema: &mut Value) {
        let defs = match schema
            .get_mut("definitions")
            .and_then(|d| d.as_object_mut())
        {
            Some(d) => d,
            None => return,
        };

        let substitutions: Vec<(String, String)> = defs
            .keys()
            .filter(|k| sanitize_def_name(k) != **k)
            .map(|k| (k.clone(), sanitize_def_name(k)))
            .collect();

        if substitutions.is_empty() {
            return;
        }

        for (orig, sanitized) in &substitutions {
            if let Some(val) = defs.remove(orig.as_str()) {
                defs.insert(sanitized.clone(), val);
            }
        }

        rewrite_refs(schema);
    }

    static VL_SCHEMA: LazyLock<jsonschema::Validator> = LazyLock::new(|| {
        let mut schema: Value =
            serde_json::from_str(include_str!("schema/v6.json")).expect("invalid schema JSON");
        sanitize_vegalite_schema(&mut schema);
        jsonschema::options()
            .with_draft(jsonschema::Draft::Draft7)
            .should_validate_formats(false)
            .build(&schema)
            .expect("invalid JSON Schema")
    });

    fn assert_valid_vegalite(json_str: &str) {
        let spec: Value = serde_json::from_str(json_str).expect("invalid JSON");
        let errors: Vec<String> = VL_SCHEMA
            .iter_errors(&spec)
            .map(|e| format!("  - {} (at {})", e, e.instance_path()))
            .collect();
        if !errors.is_empty() {
            panic!(
                "Vega-Lite schema validation failed ({} errors):\n{}",
                errors.len(),
                errors.join("\n")
            );
        }
    }

    /// Helper to wrap a DataFrame in a data map for testing (uses layer 0 key)
    fn wrap_data(df: DataFrame) -> HashMap<String, DataFrame> {
        wrap_data_for_layers(df, 1)
    }

    /// Helper to wrap a DataFrame for multiple layers (clones for each layer)
    fn wrap_data_for_layers(df: DataFrame, num_layers: usize) -> HashMap<String, DataFrame> {
        let mut data_map = HashMap::new();
        for i in 0..num_layers {
            data_map.insert(naming::layer_key(i), df.clone());
        }
        data_map
    }

    /// Helper to transform a spec's aesthetics to internal names (simulates what parser does)
    fn transform_spec(spec: &mut Plot) {
        spec.initialize_aesthetic_context();
        spec.transform_aesthetics_to_internal();
    }

    /// Helper to build a layer with pos1 and pos2 aesthetics already set up
    ///
    /// By default, maps "x" column to pos1 aesthetic and "y" column to pos2 aesthetic.
    /// Additional aesthetics and parameters can be added via builder methods.
    ///
    /// # Example
    /// ```
    /// let layer = build_layer(Geom::point())
    ///     .with_aesthetic("color".to_string(), AestheticValue::standard_column("category".to_string()));
    /// ```
    fn build_layer(geom: Geom) -> Layer {
        Layer::new(geom)
            .with_aesthetic(
                "pos1".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "pos2".to_string(),
                AestheticValue::standard_column("y".to_string()),
            )
    }

    /// Helper to build a complete spec with a single layer
    ///
    /// Creates a Plot with one layer that has pos1 and pos2 aesthetics mapped to "x" and "y" columns.
    /// Additional aesthetics and parameters can be added to the layer before calling this.
    ///
    /// # Example
    /// ```
    /// let spec = build_spec(Geom::line());
    /// ```
    fn build_spec(geom: Geom) -> Plot {
        let mut spec = Plot::new();
        let mut layer = build_layer(geom);
        // Resolve aesthetics (normally done in execution pipeline)
        layer.resolve_aesthetics();
        spec.layers.push(layer);
        spec
    }

    /// Helper to create a simple DataFrame with x and y columns for testing
    fn simple_df() -> DataFrame {
        df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
        }
        .unwrap()
    }

    #[test]
    fn test_geom_to_mark_mapping() {
        // All marks should be objects with type and clip: true
        assert_eq!(
            geom_to_mark(&Geom::point()),
            json!({"type": "point", "clip": true})
        );
        assert_eq!(
            geom_to_mark(&Geom::line()),
            json!({"type": "line", "clip": true})
        );
        assert_eq!(
            geom_to_mark(&Geom::bar()),
            json!({"type": "bar", "clip": true})
        );
        assert_eq!(
            geom_to_mark(&Geom::area()),
            json!({"type": "area", "clip": true})
        );
    }

    #[test]
    fn test_aesthetic_name_mapping() {
        use crate::plot::AestheticContext;

        use crate::plot::projection::Projection;

        // Test with cartesian projection (None = default cartesian)
        let ctx = AestheticContext::from_static(&["x", "y"], &[]);
        let cartesian = get_projection_renderer(None, None, &[]);
        let cart = cartesian.as_ref();

        // Internal position names should map to Vega-Lite channel names based on projection
        assert_eq!(map_aesthetic_name("pos1", &ctx, cart), "x");
        assert_eq!(map_aesthetic_name("pos2", &ctx, cart), "y");
        assert_eq!(map_aesthetic_name("pos1end", &ctx, cart), "x2");
        assert_eq!(map_aesthetic_name("pos2end", &ctx, cart), "y2");

        // Material aesthetics pass through directly
        assert_eq!(map_aesthetic_name("color", &ctx, cart), "color");
        assert_eq!(map_aesthetic_name("fill", &ctx, cart), "fill");
        assert_eq!(map_aesthetic_name("stroke", &ctx, cart), "stroke");
        assert_eq!(map_aesthetic_name("opacity", &ctx, cart), "opacity");
        assert_eq!(map_aesthetic_name("size", &ctx, cart), "size");
        assert_eq!(map_aesthetic_name("shape", &ctx, cart), "shape");

        // Other mapped aesthetics
        assert_eq!(map_aesthetic_name("linetype", &ctx, cart), "strokeDash");
        assert_eq!(map_aesthetic_name("linewidth", &ctx, cart), "strokeWidth");
        assert_eq!(map_aesthetic_name("label", &ctx, cart), "text");
        assert_eq!(map_aesthetic_name("fontsize", &ctx, cart), "size");

        // Test with polar projection - internal position maps to radius/theta
        // regardless of the context's user-facing names
        let polar_proj = Projection::polar();
        let polar = get_projection_renderer(Some(&polar_proj), None, &[]);
        let pol = polar.as_ref();

        let polar_ctx = AestheticContext::from_static(&["radius", "theta"], &[]);
        assert_eq!(map_aesthetic_name("pos1", &polar_ctx, pol), "radius");
        assert_eq!(map_aesthetic_name("pos2", &polar_ctx, pol), "theta");
        assert_eq!(map_aesthetic_name("pos1end", &polar_ctx, pol), "radius2");
        assert_eq!(map_aesthetic_name("pos2end", &polar_ctx, pol), "theta2");

        // Even with custom position names (e.g., PROJECT y, x TO polar),
        // internal pos1/pos2 should still map to radius/theta for Vega-Lite
        let custom_ctx = AestheticContext::from_static(&["y", "x"], &[]);
        assert_eq!(map_aesthetic_name("pos1", &custom_ctx, pol), "radius");
        assert_eq!(map_aesthetic_name("pos2", &custom_ctx, pol), "theta");
    }

    #[test]
    fn test_validation_requires_layers() {
        let writer = VegaLiteWriter::new();
        let spec = Plot::new();
        assert!(writer.validate(&spec).is_err());
    }

    #[test]
    fn test_simple_point_spec() {
        let writer = VegaLiteWriter::new();

        // Create a simple spec with user-facing aesthetic names
        let mut spec = Plot::new();
        let layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            );
        spec.layers.push(layer);

        // Create simple DataFrame
        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
        }
        .unwrap();

        // Generate Vega-Lite JSON
        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        // Verify structure (uses layer array with inline data)
        assert_eq!(vl_spec["$schema"], writer.schema);
        assert!(vl_spec["layer"].is_array());
        assert_eq!(vl_spec["layer"][0]["mark"]["type"], "point");
        assert_eq!(vl_spec["layer"][0]["mark"]["clip"], true);
        assert!(vl_spec["data"]["values"].is_array());
        assert_eq!(vl_spec["data"]["values"].as_array().unwrap().len(), 3);
        assert!(vl_spec["layer"][0]["encoding"]["x"].is_object());
        assert!(vl_spec["layer"][0]["encoding"]["y"].is_object());
    }

    #[test]
    fn test_with_title() {
        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();
        let layer = Layer::new(Geom::line())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("date".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("value".to_string()),
            );
        spec.layers.push(layer);

        let mut labels = Labels {
            labels: HashMap::new(),
        };
        labels
            .labels
            .insert("title".to_string(), Some("My Chart".to_string()));
        spec.labels = Some(labels);

        let df = df! {
            "date" => vec!["2024-01-01", "2024-01-02"],
            "value" => vec![10, 20],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        assert_eq!(vl_spec["title"], "My Chart");
        assert_eq!(vl_spec["layer"][0]["mark"]["type"], "line");
        assert_eq!(vl_spec["layer"][0]["mark"]["clip"], true);
    }

    #[test]
    fn test_labels_newline_splitting() {
        use crate::execute;
        use crate::reader::DuckDBReader;

        // Test that LABEL clause values with newlines are split into arrays

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

        let query = r#"
            SELECT
                n::INTEGER as x,
                n::INTEGER as y,
                CASE WHEN n % 2 = 0 THEN 'Group A' ELSE 'Group B' END as category
            FROM generate_series(0, 2) as t(n)
            VISUALISE x, y, category AS stroke
            DRAW point
            LABEL
                title => 'Multi-line\nChart Title',
                subtitle => 'Line 1\nLine 2\nLine 3',
                x => 'X Axis\nWith Newline',
                y => 'Single Line',
                stroke => 'Category\nMulti-line Legend'
        "#;

        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: Value = serde_json::from_str(&json_str).unwrap();

        // Check title (should be object with text and subtitle)
        assert!(vl_spec["title"].is_object(), "Title should be an object");
        let title_obj = vl_spec["title"].as_object().unwrap();

        // Check title text (multi-line, should be array)
        assert!(
            title_obj["text"].is_array(),
            "Title with newline should be an array"
        );
        let title_lines = title_obj["text"].as_array().unwrap();
        assert_eq!(title_lines.len(), 2);
        assert_eq!(title_lines[0].as_str().unwrap(), "Multi-line");
        assert_eq!(title_lines[1].as_str().unwrap(), "Chart Title");

        // Check subtitle (multi-line, should be array)
        assert!(
            title_obj["subtitle"].is_array(),
            "Subtitle with newlines should be an array"
        );
        let subtitle_lines = title_obj["subtitle"].as_array().unwrap();
        assert_eq!(subtitle_lines.len(), 3);
        assert_eq!(subtitle_lines[0].as_str().unwrap(), "Line 1");
        assert_eq!(subtitle_lines[1].as_str().unwrap(), "Line 2");
        assert_eq!(subtitle_lines[2].as_str().unwrap(), "Line 3");

        // Check x axis label (multi-line, should be array)
        let x_encoding = &vl_spec["layer"][0]["encoding"]["x"];
        assert!(
            x_encoding["title"].is_array(),
            "X axis label with newline should be an array"
        );
        let x_label_lines = x_encoding["title"].as_array().unwrap();
        assert_eq!(x_label_lines.len(), 2);
        assert_eq!(x_label_lines[0].as_str().unwrap(), "X Axis");
        assert_eq!(x_label_lines[1].as_str().unwrap(), "With Newline");

        // Check y axis label (single line, should be string)
        let y_encoding = &vl_spec["layer"][0]["encoding"]["y"];
        assert!(
            y_encoding["title"].is_string(),
            "Y axis label without newline should be a string"
        );
        assert_eq!(y_encoding["title"].as_str().unwrap(), "Single Line");

        // Check stroke legend label (multi-line, should be array)
        let stroke_encoding = &vl_spec["layer"][0]["encoding"]["stroke"];
        assert!(
            stroke_encoding["title"].is_array(),
            "Stroke legend title with newline should be an array"
        );
        let stroke_label_lines = stroke_encoding["title"].as_array().unwrap();
        assert_eq!(stroke_label_lines.len(), 2);
        assert_eq!(stroke_label_lines[0].as_str().unwrap(), "Category");
        assert_eq!(stroke_label_lines[1].as_str().unwrap(), "Multi-line Legend");
    }

    #[test]
    fn test_fontsize_linear_scaling() {
        use crate::plot::{ArrayElement, OutputRange, Scale, ScaleType};

        let writer = VegaLiteWriter::new();

        // Create spec with text geom using fontsize aesthetic
        let mut spec = Plot::new();
        let layer = Layer::new(Geom::text())
            .with_aesthetic(
                "pos1".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "pos2".to_string(),
                AestheticValue::standard_column("y".to_string()),
            )
            .with_aesthetic(
                "label".to_string(),
                AestheticValue::standard_column("label".to_string()),
            )
            .with_aesthetic(
                "fontsize".to_string(),
                AestheticValue::standard_column("value".to_string()),
            );
        spec.layers.push(layer);

        // Add fontsize scale with explicit range
        let mut scale = Scale::new("fontsize");
        scale.scale_type = Some(ScaleType::continuous());
        scale.output_range = Some(OutputRange::Array(vec![
            ArrayElement::Number(10.0),
            ArrayElement::Number(20.0),
        ]));
        spec.scales.push(scale);

        // Create DataFrame
        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![1, 2, 3],
            "label" => vec!["A", "B", "C"],
            "value" => vec![1.0, 2.0, 3.0],
        }
        .unwrap();

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

        // Verify fontsize maps to size channel
        let encoding = &vl_spec["layer"][0]["encoding"];
        assert!(encoding["size"].is_object(), "Should have size encoding");
        assert!(
            encoding["fontsize"].is_null(),
            "Should not have fontsize encoding"
        );

        // Verify scale range is linear (no area conversion)
        let scale_range = &encoding["size"]["scale"]["range"];
        assert!(scale_range.is_array(), "Scale should have range array");
        let range = scale_range.as_array().unwrap();
        assert_eq!(range.len(), 2);
        // Should be 10 and 20 converted to pixels, NOT ~31 and ~126 (which would be area-converted)
        assert_eq!(range[0].as_f64().unwrap(), 10.0 * POINTS_TO_PIXELS);
        assert_eq!(range[1].as_f64().unwrap(), 20.0 * POINTS_TO_PIXELS);
    }

    #[test]
    fn test_literal_color() {
        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();
        let layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            )
            .with_aesthetic(
                "stroke".to_string(),
                AestheticValue::Literal(ParameterValue::String("red".to_string())),
            );
        spec.layers.push(layer);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        assert_eq!(vl_spec["layer"][0]["encoding"]["stroke"]["value"], "red");
    }

    #[test]
    fn test_missing_column_error() {
        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();
        let layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("nonexistent".to_string()),
            );
        spec.layers.push(layer);
        transform_spec(&mut spec);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
        }
        .unwrap();

        let result = writer.write(&spec, &wrap_data(df));
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(err.to_string().contains("nonexistent"));
        assert!(err.to_string().contains("does not exist"));
    }

    #[test]
    fn test_numeric_type_inference_integers() {
        let df = df! {
            "x" => vec![1i64, 2, 3],
        }
        .unwrap();

        assert_eq!(infer_field_type(&df, "x"), "quantitative");
    }

    #[test]
    fn test_nominal_type_inference_strings() {
        let df = df! {
            "category" => vec!["A", "B", "C"],
        }
        .unwrap();

        assert_eq!(infer_field_type(&df, "category"), "nominal");
    }

    #[test]
    fn test_numeric_string_type_inference() {
        let df = df! {
            "numbers_as_strings" => vec!["1.5", "2.5", "3.5"],
        }
        .unwrap();

        assert_eq!(infer_field_type(&df, "numbers_as_strings"), "quantitative");
    }

    #[test]
    fn test_find_bin_for_value() {
        let breaks = vec![0.0, 10.0, 20.0, 30.0];

        // Test value in first bin [0, 10)
        assert_eq!(find_bin_for_value(5.0, &breaks), Some((0.0, 10.0)));

        // Test value at bin boundary (belongs to next bin due to half-open interval)
        assert_eq!(find_bin_for_value(10.0, &breaks), Some((10.0, 20.0)));

        // Test value in middle bin [10, 20)
        assert_eq!(find_bin_for_value(15.0, &breaks), Some((10.0, 20.0)));

        // Test value in last bin [20, 30] (closed interval)
        assert_eq!(find_bin_for_value(25.0, &breaks), Some((20.0, 30.0)));

        // Test value at last edge (should be included in last bin)
        assert_eq!(find_bin_for_value(30.0, &breaks), Some((20.0, 30.0)));

        // Test value outside range
        assert_eq!(find_bin_for_value(-5.0, &breaks), None);
        assert_eq!(find_bin_for_value(35.0, &breaks), None);

        // Test with too few breaks
        assert_eq!(find_bin_for_value(5.0, &[10.0]), None);
    }

    #[test]
    fn test_multi_layer_composition() {
        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();

        // Add line layer
        let line_layer = Layer::new(Geom::line())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            );
        spec.layers.push(line_layer);

        // Add point layer
        let point_layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            );
        spec.layers.push(point_layer);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data_for_layers(df, 2)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        // Should have layer array with 2 layers
        assert!(vl_spec["layer"].is_array());
        assert_eq!(vl_spec["layer"].as_array().unwrap().len(), 2);
        assert_eq!(vl_spec["layer"][0]["mark"]["type"], "line");
        assert_eq!(vl_spec["layer"][1]["mark"]["type"], "point");
    }

    #[test]
    fn test_build_symbol_legend_label_mapping_basic() {
        // Test the build_symbol_legend_label_mapping function directly
        use super::encoding::build_symbol_legend_label_mapping;

        let breaks = vec![
            ArrayElement::Number(0.0),
            ArrayElement::Number(25.0),
            ArrayElement::Number(50.0),
            ArrayElement::Number(75.0),
            ArrayElement::Number(100.0),
        ];

        let mut label_mapping = HashMap::new();
        label_mapping.insert("0".to_string(), Some("Low".to_string()));
        label_mapping.insert("25".to_string(), Some("Medium".to_string()));
        label_mapping.insert("50".to_string(), Some("High".to_string()));
        label_mapping.insert("75".to_string(), Some("Very High".to_string()));
        label_mapping.insert("100".to_string(), Some("Max".to_string())); // Will be excluded

        let result = build_symbol_legend_label_mapping(&breaks, &label_mapping, "left");

        // VL generates: "0 – 25", "25 – 50", "50 – 75", "≥ 75"
        // We map to range format using custom labels: "lower_label – upper_label"
        assert_eq!(
            result.get("0 – 25"),
            Some(&Some("Low – Medium".to_string()))
        );
        assert_eq!(
            result.get("25 – 50"),
            Some(&Some("Medium – High".to_string()))
        );
        assert_eq!(
            result.get("50 – 75"),
            Some(&Some("High – Very High".to_string()))
        );
        assert_eq!(
            result.get("≥ 75"),
            Some(&Some("Very High – Max".to_string()))
        );

        // Should not include a mapping for the last terminal value directly
        assert!(!result.contains_key("100"));
    }

    #[test]
    fn test_symbol_legend_label_expr_uses_range_format() {
        // Test that symbol legend labelExpr maps VL's range labels to our labels
        use crate::plot::scale::Scale;
        use crate::plot::{ArrayElement, ParameterValue, ScaleType};

        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();
        let layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            )
            .with_aesthetic(
                "size".to_string(),
                AestheticValue::standard_column("value".to_string()),
            );
        spec.layers.push(layer);

        // Add binned size scale (symbol legend case)
        let mut scale = Scale::new("size");
        scale.scale_type = Some(ScaleType::binned());
        scale.properties.insert(
            "breaks".to_string(),
            ParameterValue::Array(vec![
                ArrayElement::Number(0.0),
                ArrayElement::Number(25.0),
                ArrayElement::Number(50.0),
                ArrayElement::Number(75.0),
                ArrayElement::Number(100.0),
            ]),
        );
        // Add label renaming
        let mut labels = HashMap::new();
        labels.insert("0".to_string(), Some("Low".to_string()));
        labels.insert("25".to_string(), Some("Medium".to_string()));
        labels.insert("50".to_string(), Some("High".to_string()));
        labels.insert("75".to_string(), Some("Very High".to_string()));
        scale.label_mapping = Some(labels);
        spec.scales.push(scale);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![10, 45, 80],
            "value" => vec![10.0, 45.0, 80.0],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        // Check that labelExpr contains VL's range-style format
        let label_expr = &vl_spec["layer"][0]["encoding"]["size"]["legend"]["labelExpr"];
        assert!(label_expr.is_string());
        let expr = label_expr.as_str().unwrap();

        // Should contain mappings for VL's range format labels to our range format
        assert!(
            expr.contains("0 – 25"),
            "labelExpr should contain VL's range format '0 – 25', got: {}",
            expr
        );
        assert!(
            expr.contains("'Low – Medium'"),
            "labelExpr should map to 'Low – Medium', got: {}",
            expr
        );
        assert!(
            expr.contains("≥ 75"),
            "labelExpr should contain VL's last bin format '≥ 75', got: {}",
            expr
        );
        // Note: last bin maps "≥ 75" to "Very High – 100" (no custom label for 100 in this test)
        assert!(
            expr.contains("'Very High"),
            "labelExpr should contain 'Very High', got: {}",
            expr
        );
    }

    #[test]
    fn test_symbol_legend_open_format_with_oob_squish() {
        // Test that oob='squish' produces open format labels for symbol legends
        use super::encoding::build_symbol_legend_label_mapping;

        let breaks = vec![
            ArrayElement::Number(0.0),
            ArrayElement::Number(25.0),
            ArrayElement::Number(50.0),
            ArrayElement::Number(75.0),
            ArrayElement::Number(100.0),
        ];

        // Suppress first and last terminals (oob='squish' behavior)
        let mut label_mapping = HashMap::new();
        label_mapping.insert("0".to_string(), None); // Suppressed
        label_mapping.insert("25".to_string(), Some("Medium".to_string()));
        label_mapping.insert("50".to_string(), Some("High".to_string()));
        label_mapping.insert("75".to_string(), Some("Very High".to_string()));
        label_mapping.insert("100".to_string(), None); // Suppressed

        // Test with closed='left' (default)
        let result_left = build_symbol_legend_label_mapping(&breaks, &label_mapping, "left");

        // First bin: suppressed lower terminal → "< 25" (open format)
        assert_eq!(
            result_left.get("0 – 25"),
            Some(&Some("< Medium".to_string())),
            "First bin with suppressed lower should use '< upper' format"
        );
        // Last bin: suppressed upper terminal → "≥ 75" (open format, same as normal)
        assert_eq!(
            result_left.get("≥ 75"),
            Some(&Some("≥ Very High".to_string())),
            "Last bin with suppressed upper should use '≥ lower' format"
        );

        // Test with closed='right'
        let result_right = build_symbol_legend_label_mapping(&breaks, &label_mapping, "right");

        // First bin: suppressed lower terminal → "≤ 25" (right-closed means upper included)
        assert_eq!(
            result_right.get("0 – 25"),
            Some(&Some("≤ Medium".to_string())),
            "First bin with closed='right' should use '≤ upper' format"
        );
        // Last bin: suppressed upper terminal → "> 75" (right-closed means lower not included)
        assert_eq!(
            result_right.get("≥ 75"),
            Some(&Some("> Very High".to_string())),
            "Last bin with closed='right' should use '> lower' format"
        );
    }

    #[test]
    fn test_default_aesthetics_applied() {
        let writer = VegaLiteWriter::new();

        // Point geom without explicit size/stroke - should use defaults
        let spec = build_spec(Geom::point());

        let result = writer.write(&spec, &wrap_data(simple_df()));
        assert!(result.is_ok());
        let json_str = result.unwrap();
        assert_valid_vegalite(&json_str);
        let json: Value = serde_json::from_str(&json_str).unwrap();

        // Single-layer spec uses layer array structure
        let encoding = &json["layer"][0]["encoding"];

        // Point default stroke = "black"
        assert_eq!(encoding["stroke"]["value"], "black");

        // Point default opacity = 0.8 (retargeted to fillOpacity when fill is present)
        assert_eq!(encoding["fillOpacity"]["value"], 0.8);
    }

    #[test]
    fn test_setting_overrides_default() {
        let writer = VegaLiteWriter::new();

        // Point with SETTING opacity => 0.5 should override default (1.0)
        let mut spec = Plot::new();
        let mut layer = build_layer(Geom::point())
            .with_parameter("opacity".to_string(), ParameterValue::Number(0.5));

        // Resolve aesthetics (normally done in execution pipeline)
        layer.resolve_aesthetics();
        spec.layers.push(layer);

        let result = writer.write(&spec, &wrap_data(simple_df()));
        assert!(result.is_ok());
        let json_str = result.unwrap();
        assert_valid_vegalite(&json_str);
        let json: Value = serde_json::from_str(&json_str).unwrap();

        let encoding = &json["layer"][0]["encoding"];

        // Should use SETTING value (0.5), not default (0.8)
        // Opacity is retargeted to fillOpacity when fill is present
        assert_eq!(encoding["fillOpacity"]["value"], 0.5);
    }

    #[test]
    fn test_mapping_overrides_default() {
        let writer = VegaLiteWriter::new();

        // Point with MAPPING stroke AS stroke should override default
        let mut spec = Plot::new();
        let mut layer = Layer::new(Geom::point())
            .with_aesthetic(
                "pos1".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "pos2".to_string(),
                AestheticValue::standard_column("y".to_string()),
            )
            .with_aesthetic(
                "stroke".to_string(),
                AestheticValue::standard_column("stroke".to_string()),
            );

        // Resolve aesthetics (normally done in execution pipeline)
        layer.resolve_aesthetics();
        spec.layers.push(layer);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
            "stroke" => vec!["red", "blue", "green"],
        }
        .unwrap();

        let result = writer.write(&spec, &wrap_data(df));
        assert!(result.is_ok());
        let json_str = result.unwrap();
        assert_valid_vegalite(&json_str);
        let json: Value = serde_json::from_str(&json_str).unwrap();

        let encoding = &json["layer"][0]["encoding"];

        // Should have field encoding, not value encoding
        assert!(encoding["stroke"]["field"].is_string());
        assert_eq!(encoding["stroke"]["field"], "stroke");
        assert!(encoding["stroke"]["value"].is_null());
    }

    #[test]
    fn test_null_defaults_not_applied() {
        let writer = VegaLiteWriter::new();

        // Point has linetype as Null - should not appear in encoding
        let mut spec = Plot::new();
        let mut layer = Layer::new(Geom::point())
            .with_aesthetic(
                "pos1".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "pos2".to_string(),
                AestheticValue::standard_column("y".to_string()),
            );

        // Resolve aesthetics (normally done in execution pipeline)
        layer.resolve_aesthetics();
        spec.layers.push(layer);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
        }
        .unwrap();

        let result = writer.write(&spec, &wrap_data(df));
        assert!(result.is_ok());
        let json_str = result.unwrap();
        assert_valid_vegalite(&json_str);
        let json: Value = serde_json::from_str(&json_str).unwrap();

        // Point has linetype => Null, should not appear in encoding
        assert!(json["encoding"]["strokeDash"].is_null());
    }

    #[test]
    fn test_linetype_translated_to_stroke_dash() {
        let writer = VegaLiteWriter::new();

        // Line with linetype as SETTING (literal)
        let mut spec = Plot::new();
        let mut layer = build_layer(Geom::line()).with_aesthetic(
            "linetype".to_string(),
            AestheticValue::Literal(ParameterValue::String("dashed".to_string())),
        );

        // Resolve aesthetics (normally done in execution pipeline)
        layer.resolve_aesthetics();
        spec.layers.push(layer);

        let result = writer.write(&spec, &wrap_data(simple_df()));
        assert!(result.is_ok());
        let json_str = result.unwrap();
        assert_valid_vegalite(&json_str);
        let json: Value = serde_json::from_str(&json_str).unwrap();

        let encoding = &json["layer"][0]["encoding"];

        // "dashed" should translate to [6, 4]
        assert!(encoding["strokeDash"]["value"].is_array());
        assert_eq!(encoding["strokeDash"]["value"], json!([6, 4]));
    }

    #[test]
    fn test_linetype_default_translated_to_stroke_dash() {
        let writer = VegaLiteWriter::new();

        // Line geom has linetype default of "solid"
        let spec = build_spec(Geom::line());

        let result = writer.write(&spec, &wrap_data(simple_df()));
        assert!(result.is_ok());
        let json_str = result.unwrap();
        assert_valid_vegalite(&json_str);
        let json: Value = serde_json::from_str(&json_str).unwrap();

        let encoding = &json["layer"][0]["encoding"];

        // "solid" should translate to empty array []
        assert!(encoding["strokeDash"]["value"].is_array());
        assert_eq!(encoding["strokeDash"]["value"], json!([]));
    }

    #[test]
    fn test_facet_ordering_uses_input_range() {
        // Test that resolve_facet_ordering uses input_range (FROM clause) for discrete scales
        use crate::plot::scale::Scale;

        // Create a scale with input_range (simulating SCALE panel FROM ['A', 'B', 'C'])
        let mut scale = Scale::new("facet1");
        scale.input_range = Some(vec![
            ArrayElement::String("A".to_string()),
            ArrayElement::String("B".to_string()),
            ArrayElement::String("C".to_string()),
        ]);

        let index_map = resolve_facet_ordering(Some(&scale));

        // Verify index map matches input_range order
        assert_eq!(index_map.len(), 3);
        assert_eq!(index_map[0], (json!("A"), 0));
        assert_eq!(index_map[1], (json!("B"), 1));
        assert_eq!(index_map[2], (json!("C"), 2));
    }

    #[test]
    fn test_facet_ordering_with_null_in_input_range() {
        // Test that resolve_facet_ordering preserves null values in input_range
        // This is the fix for the bug where null panels appear first
        use crate::plot::scale::Scale;

        // Create a scale with input_range including null at the end
        // (simulating SCALE panel FROM ['Adelie', 'Gentoo', null])
        let mut scale = Scale::new("facet1");
        scale.input_range = Some(vec![
            ArrayElement::String("Adelie".to_string()),
            ArrayElement::String("Gentoo".to_string()),
            ArrayElement::Null,
        ]);

        let index_map = resolve_facet_ordering(Some(&scale));

        // Verify null is at index 2 (last)
        assert_eq!(index_map[0], (json!("Adelie"), 0));
        assert_eq!(index_map[1], (json!("Gentoo"), 1));
        assert_eq!(index_map[2], (Value::Null, 2));
    }

    #[test]
    fn test_facet_ordering_with_null_first_in_input_range() {
        // Test that null at the beginning of input_range gets index 0
        use crate::plot::scale::Scale;

        // Create a scale with null at the beginning
        let mut scale = Scale::new("facet1");
        scale.input_range = Some(vec![
            ArrayElement::Null,
            ArrayElement::String("Adelie".to_string()),
            ArrayElement::String("Gentoo".to_string()),
        ]);

        let index_map = resolve_facet_ordering(Some(&scale));

        // Verify null is at index 0 (first)
        assert_eq!(index_map[0], (Value::Null, 0));
        assert_eq!(index_map[1], (json!("Adelie"), 1));
        assert_eq!(index_map[2], (json!("Gentoo"), 2));
    }

    #[test]
    fn test_binned_facet_reverse_sets_descending_sort() {
        use crate::plot::scale::Scale;
        use crate::plot::{ParameterValue, ScaleType};

        let mut facet_def = json!({"field": "__ggsql_aes_facet1__", "type": "nominal"});

        let mut scale = Scale::new("facet1");
        scale.scale_type = Some(ScaleType::binned());
        scale
            .properties
            .insert("reverse".to_string(), ParameterValue::Boolean(true));

        apply_binned_facet_reverse(&mut facet_def, Some(&scale));

        assert_eq!(facet_def["sort"], json!("descending"));
    }

    #[test]
    fn test_binned_facet_reverse_noop_when_not_reversed() {
        use crate::plot::scale::Scale;
        use crate::plot::ScaleType;

        let mut facet_def = json!({"field": "__ggsql_aes_facet1__", "type": "nominal"});

        let mut scale = Scale::new("facet1");
        scale.scale_type = Some(ScaleType::binned());

        apply_binned_facet_reverse(&mut facet_def, Some(&scale));

        assert!(facet_def.get("sort").is_none());
    }

    #[test]
    fn test_binned_facet_reverse_noop_for_discrete() {
        use crate::plot::scale::Scale;

        let mut facet_def = json!({"field": "__ggsql_aes_facet1__", "type": "nominal"});

        let mut scale = Scale::new("facet1");
        scale
            .properties
            .insert("reverse".to_string(), ParameterValue::Boolean(true));

        apply_binned_facet_reverse(&mut facet_def, Some(&scale));

        // Discrete scales should not get sort: "descending" — they use index reversal
        assert!(facet_def.get("sort").is_none());
    }

    #[test]
    fn test_indexed_facet_label_expr_basic() {
        // Test that build_indexed_facet_label_expr maps integer indices to original labels
        let index_map = vec![(json!("A"), 0), (json!("B"), 1), (json!("C"), 2)];

        let expr = build_indexed_facet_label_expr(&index_map, None);

        assert_eq!(
            expr,
            "datum.value == 0 ? 'A' : datum.value == 1 ? 'B' : datum.value == 2 ? 'C' : datum.value"
        );
    }

    #[test]
    fn test_indexed_facet_label_expr_with_renaming() {
        // Test that user label_mapping (RENAMING clause) overrides the original values
        let index_map = vec![(json!("A"), 0), (json!("B"), 1), (json!("C"), 2)];

        let mut label_mapping = HashMap::new();
        label_mapping.insert("A".to_string(), Some("Alpha".to_string()));
        label_mapping.insert("C".to_string(), Some("Charlie".to_string()));

        let expr = build_indexed_facet_label_expr(&index_map, Some(&label_mapping));

        // A and C are renamed; B keeps its original label
        assert!(expr.contains("datum.value == 0 ? 'Alpha'"), "got: {}", expr);
        assert!(expr.contains("datum.value == 1 ? 'B'"), "got: {}", expr);
        assert!(
            expr.contains("datum.value == 2 ? 'Charlie'"),
            "got: {}",
            expr
        );
    }

    #[test]
    fn test_indexed_facet_label_expr_with_null() {
        // Test that null values in the index map produce correct label mapping
        let index_map = vec![(json!("Adelie"), 0), (json!("Gentoo"), 1), (Value::Null, 2)];

        let mut label_mapping = HashMap::new();
        label_mapping.insert("null".to_string(), Some("Missing".to_string()));

        let expr = build_indexed_facet_label_expr(&index_map, Some(&label_mapping));

        assert!(
            expr.contains("datum.value == 2 ? 'Missing'"),
            "got: {}",
            expr
        );
    }

    #[test]
    fn test_indexed_facet_label_expr_suppressed_label() {
        // Test that NULL in label_mapping suppresses the label (maps to empty string)
        let index_map = vec![(json!("A"), 0), (json!("B"), 1)];

        let mut label_mapping = HashMap::new();
        label_mapping.insert("A".to_string(), None);

        let expr = build_indexed_facet_label_expr(&index_map, Some(&label_mapping));

        assert!(expr.contains("datum.value == 0 ? ''"), "got: {}", expr);
        assert!(expr.contains("datum.value == 1 ? 'B'"), "got: {}", expr);
    }

    #[test]
    fn test_discrete_facet_label_expr_renames_null() {
        // Test that 'null' key in label_mapping generates correct Vega expression
        // for comparing against JSON null (not string 'null')
        let mut mappings = HashMap::new();
        mappings.insert("Adelie".to_string(), Some("Adelie Penguin".to_string()));
        mappings.insert("null".to_string(), Some("Missing".to_string()));

        let expr = build_discrete_facet_label_expr(Some(&mappings));

        // Should contain null comparison without quotes
        assert!(
            expr.contains("datum.value == null"),
            "Label expr should use 'datum.value == null' (not string), got: {}",
            expr
        );
        // Should map null to 'Missing'
        assert!(
            expr.contains("'Missing'"),
            "Label expr should contain 'Missing', got: {}",
            expr
        );
        // Should still use string comparison for non-null values
        assert!(
            expr.contains("datum.value == 'Adelie'"),
            "Label expr should use string comparison for Adelie, got: {}",
            expr
        );
    }

    #[test]
    fn test_binned_facet_label_expr_uses_range_labels() {
        // Test that binned facet labelExpr uses range-style labels "Lower – Upper"
        use crate::plot::scale::Scale;
        use crate::plot::{ParameterValue, ScaleType};

        // Create a binned scale with breaks [0, 20, 40, 60]
        let mut scale = Scale::new("facet1");
        scale.scale_type = Some(ScaleType::binned());
        scale.properties.insert(
            "breaks".to_string(),
            ParameterValue::Array(vec![
                ArrayElement::Number(0.0),
                ArrayElement::Number(20.0),
                ArrayElement::Number(40.0),
                ArrayElement::Number(60.0),
            ]),
        );

        // Create label mapping keyed by lower bound
        let mut label_mapping = HashMap::new();
        label_mapping.insert("0".to_string(), Some("Low".to_string()));
        label_mapping.insert("20".to_string(), Some("Medium".to_string()));
        label_mapping.insert("40".to_string(), Some("High".to_string()));
        label_mapping.insert("60".to_string(), Some("Very High".to_string()));

        let expr = build_binned_facet_label_expr(Some(&label_mapping), Some(&scale));

        // Should contain midpoint comparisons:
        // Bin [0, 20) -> midpoint 10
        // Bin [20, 40) -> midpoint 30
        // Bin [40, 60] -> midpoint 50
        assert!(
            expr.contains("datum.value == 10"),
            "labelExpr should compare against midpoint 10, got: {}",
            expr
        );
        assert!(
            expr.contains("datum.value == 30"),
            "labelExpr should compare against midpoint 30, got: {}",
            expr
        );
        assert!(
            expr.contains("datum.value == 50"),
            "labelExpr should compare against midpoint 50, got: {}",
            expr
        );

        // Should map to range-style labels using custom label names
        assert!(
            expr.contains("'Low – Medium'"),
            "labelExpr should contain range label 'Low – Medium', got: {}",
            expr
        );
        assert!(
            expr.contains("'Medium – High'"),
            "labelExpr should contain range label 'Medium – High', got: {}",
            expr
        );
        assert!(
            expr.contains("'High – Very High'"),
            "labelExpr should contain range label 'High – Very High', got: {}",
            expr
        );
    }

    #[test]
    fn test_binned_facet_label_expr_with_suppressed_lower_terminal() {
        // Test that suppressed lower terminal creates open-format label "< Upper"
        use crate::plot::scale::Scale;
        use crate::plot::{ParameterValue, ScaleType};

        let mut scale = Scale::new("facet1");
        scale.scale_type = Some(ScaleType::binned());
        scale.properties.insert(
            "breaks".to_string(),
            ParameterValue::Array(vec![
                ArrayElement::Number(0.0),
                ArrayElement::Number(50.0),
                ArrayElement::Number(100.0),
            ]),
        );

        // Create label mapping with suppressed first terminal (oob='squish' behavior)
        let mut label_mapping = HashMap::new();
        label_mapping.insert("0".to_string(), None); // Suppress lower terminal
        label_mapping.insert("50".to_string(), Some("High".to_string()));
        label_mapping.insert("100".to_string(), Some("Max".to_string()));

        let expr = build_binned_facet_label_expr(Some(&label_mapping), Some(&scale));

        // First bin with suppressed lower terminal → open format "< 50" or "< High"
        // (uses upper bound label since lower is suppressed)
        assert!(
            expr.contains("'< High'"),
            "First bin with suppressed lower should use '< Upper' format, got: {}",
            expr
        );
        // Second bin should use range format
        assert!(
            expr.contains("'High – Max'"),
            "Second bin should use range format 'High – Max', got: {}",
            expr
        );
    }

    #[test]
    fn test_binned_facet_label_expr_default_range_format() {
        // Test that binned facet without label_mapping uses default range format
        use crate::plot::scale::Scale;
        use crate::plot::{ParameterValue, ScaleType};

        let mut scale = Scale::new("facet1");
        scale.scale_type = Some(ScaleType::binned());
        scale.properties.insert(
            "breaks".to_string(),
            ParameterValue::Array(vec![
                ArrayElement::Number(0.0),
                ArrayElement::Number(25.0),
                ArrayElement::Number(50.0),
            ]),
        );

        // No label_mapping - should use break values in range format
        let expr = build_binned_facet_label_expr(None, Some(&scale));

        // Should use default range format with break values
        assert!(
            expr.contains("'0 – 25'"),
            "Should use default range format '0 – 25', got: {}",
            expr
        );
        assert!(
            expr.contains("'25 – 50'"),
            "Should use default range format '25 – 50', got: {}",
            expr
        );
    }

    #[test]
    fn test_facet_free_scales_omits_domain() {
        // Test that FACET with free => [true, true] does not set explicit domains
        // This allows Vega-Lite to compute independent domains per facet panel
        use crate::plot::scale::Scale;
        use crate::plot::{ArrayElement, Facet, FacetLayout, ParameterValue};

        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();
        let layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            );
        spec.layers.push(layer);

        // Add facet with free => [true, true] (both x and y free)
        // This is the normalized format after facet resolution
        let mut facet_properties = Parameters::new();
        facet_properties.insert(
            "free".to_string(),
            ParameterValue::Array(vec![
                ArrayElement::Boolean(true), // pos1 (x) is free
                ArrayElement::Boolean(true), // pos2 (y) is free
            ]),
        );
        spec.facet = Some(Facet {
            layout: FacetLayout::Wrap {
                variables: vec!["category".to_string()],
            },
            properties: facet_properties,
            resolved: true,
        });

        // Add scale with explicit domain that should be skipped
        let mut x_scale = Scale::new("x");
        x_scale.input_range = Some(vec![ArrayElement::Number(0.0), ArrayElement::Number(100.0)]);
        spec.scales.push(x_scale);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
            "category" => vec!["A", "A", "B"],
            "__ggsql_aes_facet1__" => vec!["A", "A", "B"],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        // Verify resolve.scale is set to independent for both axes
        assert_eq!(
            vl_spec["resolve"]["scale"]["x"], "independent",
            "x scale should be independent"
        );
        assert_eq!(
            vl_spec["resolve"]["scale"]["y"], "independent",
            "y scale should be independent"
        );

        // Verify NO explicit domain is set on x encoding (would override free scales)
        // The encoding should exist but scale.domain should not be present
        let x_encoding = &vl_spec["spec"]["layer"][0]["encoding"]["x"];
        let has_domain = x_encoding
            .get("scale")
            .and_then(|s| s.get("domain"))
            .is_some();
        assert!(
            !has_domain,
            "x encoding should NOT have explicit domain when using free scales, got: {}",
            serde_json::to_string_pretty(&x_encoding).unwrap()
        );
    }

    #[test]
    fn test_facet_free_y_only_omits_y_domain() {
        // Test that FACET with free => [false, true] omits y domain but keeps x domain
        use crate::plot::scale::Scale;
        use crate::plot::{ArrayElement, Facet, FacetLayout, ParameterValue};

        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();
        let layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            );
        spec.layers.push(layer);

        // Add facet with free => [false, true] (only y is free)
        // This is the normalized format after facet resolution
        let mut facet_properties = Parameters::new();
        facet_properties.insert(
            "free".to_string(),
            ParameterValue::Array(vec![
                ArrayElement::Boolean(false), // pos1 (x) is fixed
                ArrayElement::Boolean(true),  // pos2 (y) is free
            ]),
        );
        spec.facet = Some(Facet {
            layout: FacetLayout::Wrap {
                variables: vec!["category".to_string()],
            },
            properties: facet_properties,
            resolved: true,
        });

        // Add scales with explicit domains
        let mut x_scale = Scale::new("x");
        x_scale.input_range = Some(vec![ArrayElement::Number(0.0), ArrayElement::Number(100.0)]);
        spec.scales.push(x_scale);

        let mut y_scale = Scale::new("y");
        y_scale.input_range = Some(vec![ArrayElement::Number(0.0), ArrayElement::Number(50.0)]);
        spec.scales.push(y_scale);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
            "category" => vec!["A", "A", "B"],
            "__ggsql_aes_facet1__" => vec!["A", "A", "B"],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        // Verify only y scale is independent
        assert!(
            vl_spec["resolve"]["scale"].get("x").is_none(),
            "x scale should not be in resolve (shared)"
        );
        assert_eq!(
            vl_spec["resolve"]["scale"]["y"], "independent",
            "y scale should be independent"
        );

        // x encoding SHOULD have domain (not free)
        let x_encoding = &vl_spec["spec"]["layer"][0]["encoding"]["x"];
        let x_has_domain = x_encoding
            .get("scale")
            .and_then(|s| s.get("domain"))
            .is_some();
        assert!(
            x_has_domain,
            "x encoding SHOULD have domain when using free => 'y'"
        );

        // y encoding should NOT have domain (free)
        let y_encoding = &vl_spec["spec"]["layer"][0]["encoding"]["y"];
        let y_has_domain = y_encoding
            .get("scale")
            .and_then(|s| s.get("domain"))
            .is_some();
        assert!(
            !y_has_domain,
            "y encoding should NOT have domain when using free => 'y', got: {}",
            serde_json::to_string_pretty(&y_encoding).unwrap()
        );
    }

    #[test]
    fn test_facet_fixed_scales_keeps_domain() {
        // Test that FACET without free property (default) keeps explicit domains
        use crate::plot::scale::Scale;
        use crate::plot::{ArrayElement, Facet, FacetLayout};

        let writer = VegaLiteWriter::new();

        let mut spec = Plot::new();
        let layer = Layer::new(Geom::point())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y".to_string()),
            );
        spec.layers.push(layer);

        // Add facet without free property (default = fixed/shared scales)
        spec.facet = Some(Facet {
            layout: FacetLayout::Wrap {
                variables: vec!["category".to_string()],
            },
            properties: Parameters::new(), // No free property
            resolved: true,
        });

        // Add scale with explicit domain
        let mut x_scale = Scale::new("x");
        x_scale.input_range = Some(vec![ArrayElement::Number(0.0), ArrayElement::Number(100.0)]);
        spec.scales.push(x_scale);

        let df = df! {
            "x" => vec![1, 2, 3],
            "y" => vec![4, 5, 6],
            "category" => vec!["A", "A", "B"],
            "__ggsql_aes_facet1__" => vec!["A", "A", "B"],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        // Verify NO resolve.scale (fixed scales don't need it)
        assert!(
            vl_spec.get("resolve").is_none(),
            "Fixed scales should not have resolve property"
        );

        // x encoding SHOULD have domain (fixed scales keep explicit domains)
        let x_encoding = &vl_spec["spec"]["layer"][0]["encoding"]["x"];
        let has_domain = x_encoding
            .get("scale")
            .and_then(|s| s.get("domain"))
            .is_some();
        assert!(
            has_domain,
            "x encoding SHOULD have domain when using fixed scales"
        );
    }

    #[test]
    fn test_schema_validation_catches_invalid_spec() {
        let writer = VegaLiteWriter::new();
        let mut spec = build_spec(Geom::point());
        let df = simple_df();
        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        assert_valid_vegalite(&json_str);

        let invalid = r#"{"$schema": "https://vega.github.io/schema/vega-lite/v6.json", "mark": "not_a_mark"}"#;
        let result = std::panic::catch_unwind(|| assert_valid_vegalite(invalid));
        assert!(result.is_err(), "invalid spec should fail validation");
    }

    #[test]
    fn test_vendored_schema_matches_upstream() {
        let response = match ureq::get(VEGALITE_SCHEMA).call() {
            Ok(r) => r,
            Err(_) => {
                eprintln!("Skipping vendored schema check: could not reach {VEGALITE_SCHEMA}");
                return;
            }
        };
        let body = response.into_body().read_to_string().unwrap();
        let upstream: Value = serde_json::from_str(&body).unwrap();
        let vendored: Value =
            serde_json::from_str(include_str!("schema/v6.json")).expect("invalid schema JSON");
        assert_eq!(
            upstream, vendored,
            "Vendored schema does not match upstream at {VEGALITE_SCHEMA}. \
             Re-download with: curl -sL '{VEGALITE_SCHEMA}' > src/writer/vegalite/schema/v6.json",
        );
    }

    #[test]
    fn test_secondary_channels_have_no_disallowed_properties() {
        // Vega-Lite secondary channels (x2, y2, theta2, radius2) only support:
        // field, aggregate, bandPosition, bin, timeUnit, title, value.
        // Properties like type, scale, and axis must NOT be emitted.
        let writer = VegaLiteWriter::new();

        // Segment geom requires pos1end and pos2end as column mappings,
        // which map to x2 and y2 in Vega-Lite.
        let mut spec = Plot::new();
        let layer = Layer::new(Geom::segment())
            .with_aesthetic(
                "x".to_string(),
                AestheticValue::standard_column("x1".to_string()),
            )
            .with_aesthetic(
                "y".to_string(),
                AestheticValue::standard_column("y1".to_string()),
            )
            .with_aesthetic(
                "xend".to_string(),
                AestheticValue::standard_column("x2".to_string()),
            )
            .with_aesthetic(
                "yend".to_string(),
                AestheticValue::standard_column("y2".to_string()),
            );
        spec.layers.push(layer);

        let df = df! {
            "x1" => vec![0, 1],
            "y1" => vec![0, 1],
            "x2" => vec![1, 2],
            "y2" => vec![1, 2],
        }
        .unwrap();

        transform_spec(&mut spec);
        let json_str = writer.write(&spec, &wrap_data(df)).unwrap();
        let vl_spec: Value = serde_json::from_str(&json_str).unwrap();

        for channel in ["x2", "y2"] {
            for layer in vl_spec["layer"].as_array().unwrap() {
                if let Some(enc) = layer.get("encoding").and_then(|e| e.get(channel)) {
                    assert!(
                        enc.get("field").is_some(),
                        "{channel} should have 'field': {enc}"
                    );
                    assert!(
                        enc.get("type").is_none(),
                        "{channel} should not have 'type': {enc}"
                    );
                    assert!(
                        enc.get("scale").is_none(),
                        "{channel} should not have 'scale': {enc}"
                    );
                    assert!(
                        enc.get("axis").is_none(),
                        "{channel} should not have 'axis': {enc}"
                    );
                    assert!(
                        enc.get("stack").is_none(),
                        "{channel} should not have 'stack': {enc}"
                    );
                }
            }
        }

        // The spec must also pass Vega-Lite schema validation
        assert_valid_vegalite(&json_str);
    }

    // =========================================================================
    // Internal aesthetic names must not leak into vegalite writer error messages
    // =========================================================================

    mod writer_error_translation_tests {
        use super::*;

        #[test]
        fn validate_layer_columns_translates_pos1_to_x_under_cartesian() {
            let writer = VegaLiteWriter::new();
            let mut spec = Plot::new();
            let layer = Layer::new(Geom::point())
                .with_aesthetic(
                    "x".to_string(),
                    AestheticValue::standard_column("missing_col".to_string()),
                )
                .with_aesthetic(
                    "y".to_string(),
                    AestheticValue::standard_column("y".to_string()),
                );
            spec.layers.push(layer);

            let df = df! {
                "y" => vec![1, 2, 3],
            }
            .unwrap();

            transform_spec(&mut spec);

            let msg = match writer.write(&spec, &wrap_data(df)) {
                Err(GgsqlError::ValidationError(s)) => s,
                Err(other) => panic!("expected ValidationError, got: {}", other),
                Ok(_) => panic!("expected error, got success"),
            };

            assert_eq!(
                msg,
                "Column 'missing_col' referenced in aesthetic 'x' (layer 1 (global data)) does not exist.\nAvailable columns: y"
            );
        }

        #[test]
        fn validate_layer_columns_translates_pos1_to_angle_under_polar() {
            use crate::plot::projection::{Coord, Projection};

            let writer = VegaLiteWriter::new();
            let mut spec = Plot::new();
            spec.project = Some(Projection {
                aesthetics: vec!["angle".to_string(), "radius".to_string()],
                coord: Coord::polar(),
                properties: Parameters::new(),
                computed: Parameters::new(),
            });
            let layer = Layer::new(Geom::point())
                .with_aesthetic(
                    "angle".to_string(),
                    AestheticValue::standard_column("missing_col".to_string()),
                )
                .with_aesthetic(
                    "radius".to_string(),
                    AestheticValue::standard_column("radius".to_string()),
                );
            spec.layers.push(layer);

            let df = df! {
                "radius" => vec![1, 2, 3],
            }
            .unwrap();

            transform_spec(&mut spec);

            let msg = match writer.write(&spec, &wrap_data(df)) {
                Err(GgsqlError::ValidationError(s)) => s,
                Err(other) => panic!("expected ValidationError, got: {}", other),
                Ok(_) => panic!("expected error, got success"),
            };

            assert_eq!(
                msg,
                "Column 'missing_col' referenced in aesthetic 'angle' (layer 1 (global data)) does not exist.\nAvailable columns: radius"
            );
        }

        #[test]
        fn validate_layer_columns_translates_internal_aesthetic_column() {
            // When a mapping value is itself a wrapped internal column name
            // (e.g. `__ggsql_aes_pos1__`, used when an aesthetic refers to
            // another aesthetic's stat-produced column) and that column does
            // not exist, the error must show the user-facing aesthetic name,
            // not pos1.
            let writer = VegaLiteWriter::new();
            let mut spec = Plot::new();
            let layer = Layer::new(Geom::point())
                .with_aesthetic(
                    "x".to_string(),
                    AestheticValue::standard_column(naming::aesthetic_column("pos1")),
                )
                .with_aesthetic(
                    "y".to_string(),
                    AestheticValue::standard_column("y".to_string()),
                );
            spec.layers.push(layer);

            let df = df! {
                "y" => vec![1, 2, 3],
            }
            .unwrap();

            transform_spec(&mut spec);

            let msg = match writer.write(&spec, &wrap_data(df)) {
                Err(GgsqlError::ValidationError(s)) => s,
                Err(other) => panic!("expected ValidationError, got: {}", other),
                Ok(_) => panic!("expected error, got success"),
            };

            // Both the column reference and the aesthetic name must be in user form.
            assert!(
                !msg.contains("__ggsql_aes_"),
                "message must not mention raw column name: {}",
                msg
            );
            assert!(
                !msg.contains("'pos1'"),
                "message must not mention internal name 'pos1': {}",
                msg
            );
            assert!(
                msg.contains("Column 'x'"),
                "stripped column name should be displayed as 'x': {}",
                msg
            );
            assert!(
                msg.contains("aesthetic 'x'"),
                "aesthetic key should be displayed as 'x': {}",
                msg
            );
        }
    }

    #[test]
    #[cfg(feature = "duckdb")]
    fn test_boxplot_schema_validation() {
        use crate::reader::{DuckDBReader, Reader};

        let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap();
        reader
            .execute_sql(
                "CREATE TABLE box_data AS
                SELECT 'A' AS grp, generate_series * 1.0 AS value FROM GENERATE_SERIES(1, 10)
                UNION ALL
                SELECT 'B' AS grp, generate_series * 1.0 + 4.0 AS value FROM GENERATE_SERIES(1, 10)",
            )
            .unwrap();

        let spec = reader
            .execute("SELECT * FROM box_data VISUALISE grp AS x, value AS y DRAW boxplot")
            .unwrap();

        let writer = VegaLiteWriter::new();
        let json_str = writer.render(&spec).unwrap();
        assert_valid_vegalite(&json_str);
    }
}