mlt-core 0.9.2

//! Row-oriented "source form" for the optimizer.
//!
//! [`TileLayer`] holds one [`TileFeature`] per map feature, each owning
//! its geometry as a [`geo_types::Geometry<i32>`] and its property values as a
//! plain `Vec<PropValue>`.  This is the working form used throughout the
//! optimizer and sorting pipeline: it is cheap to clone, trivially sortable,
//! and free from any encoded/decoded duality.
//!
//! Conversion from [`TileLayer`] to [`StagedLayer`] is done via
//! [`StagedLayer::from_tile`] with pre-computed layer statistics.

use crate::decoder::{GeometryValues, PropValue, TileFeature, TileLayer};
use crate::encoder::model::{CurveParams, StagedLayer};
use crate::encoder::optimizer::{LayerStats, Presence, SharedDictRole};
use crate::encoder::{SortStrategy, StagedId, StagedProperty, StagedSharedDict};

impl StagedLayer {
    /// Construct a [`StagedLayer`] from a row-oriented [`TileLayer`] using
    /// pre-computed layer statistics and curve parameters.
    ///
    /// `curve_params` is taken as a parameter (rather than recomputed here)
    /// so a single [`TileLayer::curve_params`] scan feeds every sort trial
    /// and the encoder's dictionary builders.
    ///
    /// When `tessellate` is `true`, polygon and multi-polygon geometries have
    /// their triangulation stored alongside the geometry.
    #[must_use]
    #[hotpath::measure]
    pub fn from_tile(
        mut source: TileLayer,
        sort: SortStrategy,
        stats: &LayerStats,
        tessellate: bool,
        curve_params: CurveParams,
    ) -> Self {
        assert!(!source.features.is_empty(), "empty tile");
        source.sort(sort, curve_params);
        let mut geometry = if tessellate {
            GeometryValues::new_tessellated()
        } else {
            GeometryValues::default()
        };
        for f in &source.features {
            geometry.push_geom(&f.geometry);
        }

        let id = StagedId::from_optional_with_presence(
            source.features.iter().map(|f| f.id),
            stats.id.as_ref(),
        );

        let shared_dict_columns = shared_dict_columns(stats);
        let mut properties = Vec::with_capacity(source.property_names.len());
        for (col_idx, shared_cols) in shared_dict_columns
            .iter()
            .enumerate()
            .take(source.property_names.len())
        {
            let prop_analysis = stats
                .properties
                .get(col_idx)
                .expect("analysis matches source property columns");
            match prop_analysis.stats.shared_dict() {
                SharedDictRole::Owner(prefix) => {
                    properties.push(build_shared_dict(
                        col_idx,
                        &prefix,
                        shared_cols,
                        &source.property_names,
                        stats,
                        &mut source.features,
                    ));
                }
                SharedDictRole::Member(_) => {}
                SharedDictRole::None => {
                    if let Some(prop) = build_scalar_column(
                        std::mem::take(&mut source.property_names[col_idx]),
                        col_idx,
                        prop_analysis.presence,
                        &mut source.features,
                    ) {
                        properties.push(prop);
                    }
                }
            }
        }

        Self {
            name: source.name,
            extent: source.extent,
            id,
            geometry,
            properties,
        }
    }
}

fn shared_dict_columns(stats: &LayerStats) -> Vec<Vec<usize>> {
    let mut columns = vec![Vec::new(); stats.properties.len()];
    for (col_idx, prop) in stats.properties.iter().enumerate() {
        match prop.stats.shared_dict() {
            SharedDictRole::Owner(_) => columns[col_idx].push(col_idx),
            SharedDictRole::Member(owner_col) => columns[owner_col].push(col_idx),
            SharedDictRole::None => {}
        }
    }
    columns
}

fn build_scalar_column(
    name: String,
    col: usize,
    presence: Presence,
    features: &mut [TileFeature],
) -> Option<StagedProperty> {
    if presence == Presence::AllNull {
        return None;
    }

    // Determine the variant by peeking at the first feature value.
    // Typed nulls (e.g. `PropValue::Bool(None)`) already carry the column type,
    // so no filtering is needed; only a fully-absent column returns `None` here.
    // Fall back to `Str` if every feature has no value for this column.
    let first_val = features.iter().find_map(|f| f.properties.get(col));

    // Presence is precomputed before sort trials; this pass only gathers values
    // in the selected row order.
    macro_rules! scalar_col {
        ($opt_ctor:ident, $non_opt_ctor:ident, $ty:ty, $sv:ident) => {{
            Some(match presence {
                Presence::AllNull => unreachable!("handled before variant dispatch"),
                Presence::AllPresent => StagedProperty::$non_opt_ctor(
                    name,
                    features
                        .iter()
                        .map(|f| match f.properties.get(col) {
                            Some(PropValue::$sv(Some(v))) => *v,
                            _ => unreachable!("analysis guarantees present typed values"),
                        })
                        .collect(),
                ),
                Presence::Mixed | Presence::SameAsProp(_) => StagedProperty::$opt_ctor(
                    name,
                    features.iter().map(|f| match f.properties.get(col) {
                        Some(PropValue::$sv(v)) => *v,
                        _ => None,
                    }),
                ),
            })
        }};
    }

    match first_val {
        Some(PropValue::Bool(_)) => scalar_col!(opt_bool, bool, bool, Bool),
        Some(PropValue::I8(_)) => scalar_col!(opt_i8, i8, i8, I8),
        Some(PropValue::U8(_)) => scalar_col!(opt_u8, u8, u8, U8),
        Some(PropValue::I32(_)) => scalar_col!(opt_i32, i32, i32, I32),
        Some(PropValue::U32(_)) => scalar_col!(opt_u32, u32, u32, U32),
        Some(PropValue::I64(_)) => scalar_col!(opt_i64, i64, i64, I64),
        Some(PropValue::U64(_)) => scalar_col!(opt_u64, u64, u64, U64),
        Some(PropValue::F32(_)) => scalar_col!(opt_f32, f32, f32, F32),
        Some(PropValue::F64(_)) => scalar_col!(opt_f64, f64, f64, F64),
        Some(PropValue::Str(_)) | None => Some(match presence {
            Presence::AllNull => unreachable!("handled before variant dispatch"),
            Presence::AllPresent => StagedProperty::str(
                name,
                features
                    .iter_mut()
                    .map(|f| match f.properties.get_mut(col) {
                        Some(PropValue::Str(Some(v))) => std::mem::take(v),
                        _ => unreachable!("analysis guarantees present string values"),
                    }),
            ),
            Presence::Mixed | Presence::SameAsProp(_) => StagedProperty::opt_str(
                name,
                features
                    .iter_mut()
                    .map(|f| match f.properties.get_mut(col) {
                        Some(PropValue::Str(v)) => v.take(),
                        _ => None,
                    }),
            ),
        }),
    }
}

fn build_shared_dict(
    owner_col: usize,
    prefix: &str,
    shared_dict_columns: &[usize],
    property_names: &[String],
    analysis: &LayerStats,
    features: &mut [TileFeature],
) -> StagedProperty {
    debug_assert_eq!(shared_dict_columns.first(), Some(&owner_col));
    let columns = shared_dict_columns.iter().copied().map(|col_idx| {
        let name = &property_names[col_idx];
        let suffix = name.strip_prefix(prefix).unwrap_or(name).to_owned();
        let values: Vec<Option<String>> = features
            .iter_mut()
            .map(|f| match f.properties.get_mut(col_idx) {
                Some(PropValue::Str(s)) => s.take(),
                _ => None,
            })
            .collect();
        let presence = analysis.properties[col_idx].presence;
        (suffix, values, presence)
    });

    StagedProperty::SharedDict(
        StagedSharedDict::new(prefix.to_owned(), columns).expect("StagedSharedDict succeed"),
    )
}

#[cfg(test)]
mod tests {
    use geo_types::Point;

    use super::*;
    use crate::Layer;
    use crate::decoder::GeometryValues;
    use crate::encoder::{Codecs, Encoder};
    use crate::test_helpers::{dec, parser};

    fn layer_tile(staged: StagedLayer) -> TileLayer {
        let mut codecs = Codecs::default();
        let buf = staged
            .encode_into(Encoder::default(), &mut codecs)
            .unwrap()
            .into_layer_bytes()
            .unwrap();
        let (_, layer) = Layer::from_bytes(&buf, &mut parser()).unwrap();
        let Layer::Tag01(lazy) = layer else { panic!() };
        let mut d = dec();
        lazy.decode_all(&mut d).unwrap().into_tile(&mut d).unwrap()
    }

    fn two_points() -> GeometryValues {
        let mut g = GeometryValues::default();
        g.push_geom(&geo_types::Geometry::<i32>::Point(Point::new(0, 0)));
        g.push_geom(&geo_types::Geometry::<i32>::Point(Point::new(1, 1)));
        g
    }

    /// `into_tile` must produce a **typed** null (e.g. `PropValue::Bool(None)`)
    /// for null slots, matching the column's actual type, even when the **first**
    /// feature is null.
    #[test]
    fn null_first_feature_preserves_later_typed_value() {
        let tile = layer_tile(StagedLayer {
            name: "t".into(),
            extent: 4096,
            id: StagedId::None,
            geometry: two_points(),
            properties: vec![StagedProperty::opt_bool("flag", vec![None, Some(false)])],
        });

        assert_eq!(tile.property_names, vec!["flag"]);
        // Null slot → typed null matching the column type
        assert_eq!(tile.features[0].properties[0], PropValue::Bool(None));
        // Non-null value after the null must not be dropped
        assert_eq!(tile.features[1].properties[0], PropValue::Bool(Some(false)));
    }

    /// Every scalar type must produce a typed null for null slots and a typed
    /// non-null value for present slots, even when the first feature is null.
    #[test]
    fn null_first_feature_across_types() {
        let props = vec![
            StagedProperty::opt_bool("b", vec![None, Some(true)]),
            StagedProperty::opt_i8("i8", vec![None, Some(-1)]),
            StagedProperty::opt_u8("u8", vec![None, Some(2)]),
            StagedProperty::opt_i32("i32", vec![None, Some(-3)]),
            StagedProperty::opt_u32("u32", vec![None, Some(4)]),
            StagedProperty::opt_i64("i64", vec![None, Some(-5)]),
            StagedProperty::opt_u64("u64", vec![None, Some(6)]),
            StagedProperty::opt_f32("f32", vec![None, Some(7.0)]),
            StagedProperty::opt_f64("f64", vec![None, Some(8.0)]),
            StagedProperty::opt_str("s", vec![None, Some("ok")]),
        ];
        let tile = layer_tile(StagedLayer {
            name: "t".into(),
            extent: 4096,
            id: StagedId::None,
            geometry: two_points(),
            properties: props,
        });

        // Feature 0: every column is null → typed null for each column
        let n = &tile.features[0].properties;
        assert_eq!(n[0], PropValue::Bool(None));
        assert_eq!(n[1], PropValue::I8(None));
        assert_eq!(n[2], PropValue::U8(None));
        assert_eq!(n[3], PropValue::I32(None));
        assert_eq!(n[4], PropValue::U32(None));
        assert_eq!(n[5], PropValue::I64(None));
        assert_eq!(n[6], PropValue::U64(None));
        assert_eq!(n[7], PropValue::F32(None));
        assert_eq!(n[8], PropValue::F64(None));
        assert_eq!(n[9], PropValue::Str(None));

        // Feature 1: every column has its typed non-null value
        let p = &tile.features[1].properties;
        assert_eq!(p[0], PropValue::Bool(Some(true)));
        assert_eq!(p[1], PropValue::I8(Some(-1)));
        assert_eq!(p[2], PropValue::U8(Some(2)));
        assert_eq!(p[3], PropValue::I32(Some(-3)));
        assert_eq!(p[4], PropValue::U32(Some(4)));
        assert_eq!(p[5], PropValue::I64(Some(-5)));
        assert_eq!(p[6], PropValue::U64(Some(6)));
        assert_eq!(p[7], PropValue::F32(Some(7.0)));
        assert_eq!(p[8], PropValue::F64(Some(8.0)));
        assert_eq!(p[9], PropValue::Str(Some("ok".into())));
    }
}