stt-build 0.4.0

//! PostGIS input source (cargo feature `postgres`).
//!
//! Reads features directly from a live PostgreSQL/PostGIS query instead of a
//! GeoParquet file, producing the exact same [`ParsedFeature`] stream the
//! GeoParquet reader emits ([`crate::input`]) — so the whole tiler/writer
//! pipeline is reused unchanged.
//!
//! The bridge is WKB: PostGIS `ST_AsEWKB(geom)` returns the same SRID-prefixed
//! EWKB blob that [`crate::input::parse_wkb_geometry`] already decodes (via
//! `geozero`). We wrap the user's table / SQL as a subquery, project the
//! geometry to a `bytea` WKB column, and read every other returned column as a
//! feature property.
//!
//! Streaming is bounded-memory via a server-side `DECLARE … CURSOR` +
//! `FETCH FORWARD <batch>` loop, mirroring [`crate::input::stream_features`].

use anyhow::{Context, Result};
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use geojson::Feature;
use postgres::types::Type;
use postgres::{Client, NoTls, Row};
use std::sync::Arc;

use crate::db_input_common::{
    apply_int_time_format, decimal_string_to_json, json_number_or_null, warn_dropped_columns,
    RowOutcome, VertexCoercions,
};
use crate::input::{
    parse_iso8601, parse_wkb_geometry, reject_negative_timestamp, InputStrictness, ParsedFeature,
    TimeFormat,
};

/// Alias the wrapped query projects the geometry into (as `bytea` EWKB). Public
/// so a dynamic tile server building its own per-tile query keeps the geometry
/// column name that [`decode_rows`] expects.
pub const WKB_ALIAS: &str = "__stt_wkb";
/// Default rows fetched per cursor round-trip.
pub const DEFAULT_BATCH_SIZE: usize = 8192;

/// What to read from PostGIS — either a table (we `SELECT *`) or an arbitrary
/// `SELECT` the caller supplies.
#[derive(Debug, Clone)]
pub enum QuerySource {
    /// A table (optionally schema-qualified, e.g. `public.trips`).
    Table(String),
    /// An arbitrary `SELECT …` statement.
    Sql(String),
}

/// Fully describes the PostGIS read: the source, which column holds geometry,
/// an optional `WHERE` filter, and optional reprojection to EPSG:4326.
#[derive(Debug, Clone)]
pub struct QuerySpec {
    pub source: QuerySource,
    pub geom_column: String,
    pub where_clause: Option<String>,
    /// When set, geometry is `ST_Transform`ed from this SRID to 4326. Omit when
    /// the source geometry is already lon/lat (EPSG:4326).
    pub reproject_from_srid: Option<i32>,
}

impl QuerySpec {
    /// Quote a SQL identifier (double-quote, escaping embedded quotes).
    fn quote_ident(name: &str) -> String {
        format!("\"{}\"", name.replace('"', "\"\""))
    }

    /// Build the wrapped query: project the geometry to an EWKB `bytea` column
    /// aliased [`WKB_ALIAS`], and pass every other column through as `q.*`.
    /// Public so an async caller (the serve path) can prepare the same
    /// projection itself and feed the resolved columns to
    /// [`property_kinds_from_columns`].
    pub fn wrapped_query(&self) -> String {
        let inner = match &self.source {
            // A bare table name may be schema-qualified; trust the operator.
            QuerySource::Table(t) => format!("SELECT * FROM {t}"),
            QuerySource::Sql(s) => s.clone(),
        };
        let geom = format!("q.{}", Self::quote_ident(&self.geom_column));
        let wkb_expr = match self.reproject_from_srid {
            Some(srid) => {
                format!("ST_AsEWKB(ST_Transform(ST_SetSRID({geom}, {srid}), 4326))")
            }
            None => format!("ST_AsEWKB({geom})"),
        };
        let where_sql = self
            .where_clause
            .as_ref()
            .map(|w| format!(" WHERE {w}"))
            .unwrap_or_default();
        format!("SELECT {wkb_expr} AS {WKB_ALIAS}, q.* FROM ( {inner} ) AS q{where_sql}")
    }
}

/// Build a per-tile query for a dynamic server: project geometry to an EWKB
/// `bytea` aliased [`WKB_ALIAS`], pass every other column through, and filter by
/// a WGS84 bounding box (`&&` against the GiST index) and a half-open time
/// window `[t_start_ms, t_end_ms)`. The geometry / time literals are
/// server-formatted numbers (request `z/x/y/t` are parsed to integers upstream),
/// so there is no request-controlled SQL. The time column must be a
/// `timestamp`/`timestamptz` (compared via `to_timestamp`).
///
/// `source_srid`: when the stored geometry is not EPSG:4326, it is
/// `ST_Transform(ST_SetSRID(…))`-ed to 4326 *before* both the EWKB projection
/// and the bbox test (the exact expression the `--source-srid` ingest uses) —
/// the filter runs in tile space, still a strict superset. The per-row
/// transform bypasses a plain GiST index on the raw column; add a functional
/// index on the transform expression, or store 4326, if you need index speed.
///
/// Feed the result rows straight to [`decode_rows`].
#[allow(clippy::too_many_arguments)]
pub fn build_tile_query(
    table: &str,
    geom_column: &str,
    time_field: &str,
    time_format: TimeFormat,
    bbox: [f64; 4],
    t_start_ms: i64,
    t_end_ms: i64,
    source_srid: Option<i32>,
) -> String {
    let geom_raw = format!("q.{}", QuerySpec::quote_ident(geom_column));
    let geom = match source_srid {
        Some(srid) => format!("ST_Transform(ST_SetSRID({geom_raw}, {srid}), 4326)"),
        None => geom_raw,
    };
    let time = format!("q.{}", QuerySpec::quote_ident(time_field));
    let time_where = pg_time_window(&time, time_format, t_start_ms, t_end_ms);
    format!(
        "SELECT ST_AsEWKB({geom}) AS {WKB_ALIAS}, q.* \
         FROM ( SELECT * FROM {table} ) AS q \
         WHERE {geom} && ST_MakeEnvelope({}, {}, {}, {}, 4326) \
           AND {time_where}",
        bbox[0], bbox[1], bbox[2], bbox[3]
    )
}

/// The half-open `[t_start_ms, t_end_ms)` predicate for the time column, chosen
/// by `--time-format` so an INTEGER epoch column serves as well as it ingests
/// (the ingest path already honours `--time-format` for integer columns).
/// `Iso8601` (the default) assumes a `timestamp`/`timestamptz` column and
/// compares via `to_timestamp`; `UnixMs`/`UnixSec` assume an integer column and
/// compare numerically. Seconds compare the RAW column against ceil-divided
/// second bounds (`t*1000 >= a ⟺ t >= ⌈a/1000⌉` and `t*1000 < b ⟺ t <
/// ⌈b/1000⌉` for integer `t`) — exactly equivalent to scaling the column, but
/// sargable, so a b-tree index on the column still applies. Literals are
/// server-formatted numbers — no request-controlled SQL.
fn pg_time_window(time: &str, time_format: TimeFormat, t_start_ms: i64, t_end_ms: i64) -> String {
    match time_format {
        TimeFormat::Iso8601 => format!(
            "{time} >= to_timestamp({t_start_ms}::double precision / 1000.0) \
             AND {time} <  to_timestamp({t_end_ms}::double precision / 1000.0)"
        ),
        TimeFormat::UnixMs => {
            format!("{time} >= {t_start_ms} AND {time} < {t_end_ms}")
        }
        TimeFormat::UnixSec => {
            let start_s = crate::db_input_common::ceil_ms_to_seconds(t_start_ms);
            let end_s = crate::db_input_common::ceil_ms_to_seconds(t_end_ms);
            format!("{time} >= {start_s} AND {time} < {end_s}")
        }
    }
}

/// Decode a slice of PostGIS rows (from a query that projects geometry to an
/// EWKB column aliased [`WKB_ALIAS`], e.g. [`build_tile_query`]) into
/// [`ParsedFeature`]s. The per-row schema is resolved once from `rows[0]`.
#[allow(clippy::too_many_arguments)]
pub fn decode_rows(
    rows: &[Row],
    time_field: &str,
    end_time_field: Option<&str>,
    geom_column: &str,
    time_format: TimeFormat,
    time_strictness: InputStrictness,
    geometry_strictness: InputStrictness,
) -> Result<Vec<ParsedFeature>> {
    if rows.is_empty() {
        return Ok(Vec::new());
    }
    let schema = RowSchema::from_columns(
        rows[0].columns(),
        time_field,
        end_time_field,
        geom_column,
        time_format,
    )?;
    let mut out = Vec::with_capacity(rows.len());
    // Per-tile serve path: don't tally coercions (no end-of-read summary, no spam).
    let mut sink = VertexCoercions::default();
    for (i, row) in rows.iter().enumerate() {
        if let RowOutcome::Feature(f) =
            schema.parse_row(row, time_strictness, geometry_strictness, i, &mut sink)?
        {
            out.push(*f);
        }
    }
    Ok(out)
}

/// Eager variant: collect the whole PostGIS query into memory. Mirrors
/// [`crate::input::load_features`].
#[allow(clippy::too_many_arguments)]
pub fn load_features_postgres(
    conn: &str,
    spec: &QuerySpec,
    time_field: &str,
    end_time_field: Option<&str>,
    time_format: TimeFormat,
    time_strictness: InputStrictness,
    geometry_strictness: InputStrictness,
) -> Result<Vec<ParsedFeature>> {
    let mut features = Vec::new();
    stream_features_postgres(
        conn,
        spec,
        time_field,
        end_time_field,
        time_format,
        time_strictness,
        geometry_strictness,
        DEFAULT_BATCH_SIZE,
        |batch| {
            features.extend(batch);
            Ok(())
        },
    )?;
    tracing::info!("Loaded {} total features from PostGIS", features.len());
    Ok(features)
}

/// Stream a PostGIS query one cursor batch at a time, invoking `on_batch` with
/// the materialised [`ParsedFeature`]s. Peak memory is bounded by `batch_size`
/// rows, mirroring [`crate::input::stream_features`].
#[allow(clippy::too_many_arguments)]
pub fn stream_features_postgres<F>(
    conn: &str,
    spec: &QuerySpec,
    time_field: &str,
    end_time_field: Option<&str>,
    time_format: TimeFormat,
    time_strictness: InputStrictness,
    geometry_strictness: InputStrictness,
    batch_size: usize,
    mut on_batch: F,
) -> Result<()>
where
    F: FnMut(Vec<ParsedFeature>) -> Result<()>,
{
    let wrapped = spec.wrapped_query();
    tracing::debug!("PostGIS source query: {wrapped}");

    let mut client = Client::connect(conn, NoTls)
        .with_context(|| "failed to connect to PostgreSQL (NoTls; localhost / non-TLS only)")?;
    let mut txn = client.transaction()?;

    // Server-side cursor: bounded memory regardless of result size.
    let declare = format!("DECLARE stt_cur NO SCROLL CURSOR FOR {wrapped}");
    txn.batch_execute(&declare).with_context(|| {
        "failed to open PostGIS cursor — check --table/--sql, --geom-column and the connection"
    })?;
    let fetch = format!("FETCH FORWARD {} FROM stt_cur", batch_size.max(1));

    let mut schema: Option<RowSchema> = None;
    let mut total_rows = 0usize;
    let mut geom_failures = 0usize;
    // Per-vertex array elements silently defaulted (NULL/unmappable element →
    // 0 for timestamps, NaN for values), tallied once for an end-of-read summary.
    let mut vertex_coercions = VertexCoercions::default();
    // Property columns that carried a value in ANY row — so we can warn once
    // about source columns that produced nothing (unmappable type, or all-NULL).
    let mut seen_props: std::collections::HashSet<String> = std::collections::HashSet::new();

    loop {
        let rows = txn.query(&fetch, &[]).context("PostGIS cursor FETCH failed")?;
        if rows.is_empty() {
            break;
        }
        if schema.is_none() {
            schema = Some(RowSchema::from_columns(
                rows[0].columns(),
                time_field,
                end_time_field,
                &spec.geom_column,
                time_format,
            )?);
        }
        let sch = schema.as_ref().unwrap();

        let mut batch = Vec::with_capacity(rows.len());
        for (i, row) in rows.iter().enumerate() {
            match sch.parse_row(
                row,
                time_strictness,
                geometry_strictness,
                total_rows + i,
                &mut vertex_coercions,
            )? {
                RowOutcome::Feature(f) => batch.push(*f),
                RowOutcome::GeomSkip => geom_failures += 1,
            }
        }
        total_rows += rows.len();
        if total_rows % 100_000 < rows.len() {
            tracing::info!("Loaded {total_rows} rows from PostGIS...");
        }
        // Record which property columns actually produced a value (stop once all
        // have, so there's no per-row overhead in the common all-present case).
        if seen_props.len() < sch.props.len() {
            for f in &batch {
                if let Some(props) = &f.shared_properties {
                    for k in props.keys() {
                        seen_props.insert(k.clone());
                    }
                }
            }
        }
        on_batch(batch)?;
    }
    let _ = txn.batch_execute("CLOSE stt_cur");

    if geom_failures > 0 {
        tracing::warn!(
            "{geom_failures}/{total_rows} PostGIS rows had null/unparseable geometry and were skipped"
        );
    }
    if let Some(sch) = schema.as_ref() {
        warn_dropped_columns(
            sch.props
                .iter()
                .map(|p| (p.name.as_str(), Some(p.ty.name().to_string()))),
            &seen_props,
            total_rows,
            "PostGIS",
            "geometry/uuid/array",
        );
    }
    vertex_coercions.warn("PostGIS");
    Ok(())
}

/// Schema-level property-kind map for a PostGIS source — the DB analog of
/// [`crate::input::property_kinds`] (GeoParquet): prepare the wrapped query
/// (types every result column without executing it) and pin each property
/// column's tile kind from its PG type, so `TileConfig::property_types` keeps
/// the layer schema stable when a column is all-NULL within one tile (per-tile
/// value sniffing drops it there and the layer schema drifts across tiles).
/// Mirrors the DuckDB reader's `property_kinds` — the DB adaptors pin the same
/// way. Opens a short-lived connection; the ingest read reconnects afterward.
pub fn property_kinds(
    conn: &str,
    spec: &QuerySpec,
    time_field: &str,
    end_time_field: Option<&str>,
) -> Result<crate::columnar::PropertyTypes> {
    let mut client = Client::connect(conn, NoTls)
        .with_context(|| "failed to connect to PostgreSQL (NoTls; localhost / non-TLS only)")?;
    let stmt = client
        .prepare(&spec.wrapped_query())
        .context("prepare PostGIS schema probe — check --table/--sql and --geom-column")?;
    Ok(property_kinds_from_columns(
        stmt.columns(),
        time_field,
        end_time_field,
        &spec.geom_column,
    ))
}

/// As [`property_kinds`], from already-resolved result columns of any statement
/// that projects the geometry to [`WKB_ALIAS`] (the same contract as
/// [`decode_rows`]; the async serve path prepares its own statement). Excludes
/// the system columns with the same predicates [`RowSchema::from_columns`]
/// uses, so the pinned map and the row decode can't disagree about which
/// columns are properties.
pub fn property_kinds_from_columns(
    columns: &[postgres::Column],
    time_field: &str,
    end_time_field: Option<&str>,
    geom_column: &str,
) -> crate::columnar::PropertyTypes {
    let mut kinds = crate::columnar::PropertyTypes::new();
    for col in columns {
        let name = col.name();
        if name == WKB_ALIAS
            || name == time_field
            || end_time_field == Some(name)
            || name == geom_column
            || crate::input::is_coordinate_column_name(name)
            || crate::input::is_vertex_metadata_column(name)
        {
            continue;
        }
        if let Some(kind) = property_kind_for_pg(col.type_()) {
            kinds.insert(name.to_string(), kind);
        }
    }
    kinds
}

/// Schema-level mirror of [`decode_property`]: the tile kind a PG column of
/// this type decodes to. `None` = unmappable (dropped at read time), or
/// nested (`JSON`/`JSONB` decode to arbitrary JSON) — left to per-tile
/// sniffing.
fn property_kind_for_pg(ty: &Type) -> Option<crate::columnar::PropertyKind> {
    use crate::columnar::PropertyKind;
    match *ty {
        // Timestamps/dates decode to epoch-ms numbers; NUMERIC to nearest-f64.
        Type::INT2
        | Type::INT4
        | Type::INT8
        | Type::FLOAT4
        | Type::FLOAT8
        | Type::NUMERIC
        | Type::TIMESTAMP
        | Type::TIMESTAMPTZ
        | Type::DATE => Some(PropertyKind::Numeric),
        Type::BOOL | Type::TEXT | Type::VARCHAR | Type::BPCHAR | Type::NAME => {
            Some(PropertyKind::Categorical)
        }
        _ => None,
    }
}

/// A time column resolved against the result set: its index and PG type.
struct TimeCol {
    idx: usize,
    ty: Type,
}

/// A property column: index, output name, and PG type.
struct PropCol {
    idx: usize,
    name: String,
    ty: Type,
}

/// Result-set schema resolved once from the first batch's column metadata.
struct RowSchema {
    wkb_idx: usize,
    time: TimeCol,
    end_time: Option<TimeCol>,
    /// Optional per-vertex array columns, matching the GeoParquet reader's
    /// hardcoded `vertex_timestamps` / `vertex_values` / `vertex_value_matrix`
    /// columns (LineString trajectory per-vertex timing/values + animated
    /// static-geometry overviews). `TimeCol` is reused purely as an
    /// `{ idx, ty }` pair. `None` when the source query exposes no such column.
    vertex_timestamps: Option<TimeCol>,
    vertex_values: Option<TimeCol>,
    vertex_value_matrix: Option<TimeCol>,
    props: Vec<PropCol>,
    time_format: TimeFormat,
}

impl RowSchema {
    fn from_columns(
        columns: &[postgres::Column],
        time_field: &str,
        end_time_field: Option<&str>,
        geom_column: &str,
        time_format: TimeFormat,
    ) -> Result<Self> {
        let find = |name: &str| columns.iter().position(|c| c.name() == name);

        let wkb_idx = find(WKB_ALIAS).ok_or_else(|| {
            anyhow::anyhow!("internal: wrapped query did not project the {WKB_ALIAS} column")
        })?;
        let time_idx = find(time_field).ok_or_else(|| {
            anyhow::anyhow!("--time-field '{time_field}' not found in the PostGIS result columns")
        })?;
        let time = TimeCol {
            idx: time_idx,
            ty: columns[time_idx].type_().clone(),
        };
        let end_time = match end_time_field {
            Some(f) => {
                let idx = find(f).ok_or_else(|| {
                    anyhow::anyhow!("--end-time-field '{f}' not found in the PostGIS result columns")
                })?;
                Some(TimeCol {
                    idx,
                    ty: columns[idx].type_().clone(),
                })
            }
            None => None,
        };

        // Optional per-vertex array columns, detected by the exact column names
        // the GeoParquet reader hardcodes. `TimeCol` is reused as `{ idx, ty }`.
        let array_col = |name: &str| {
            find(name).map(|idx| TimeCol {
                idx,
                ty: columns[idx].type_().clone(),
            })
        };
        let vertex_timestamps = array_col("vertex_timestamps");
        let vertex_values = array_col("vertex_values");
        let vertex_value_matrix = array_col("vertex_value_matrix");

        // Every remaining column becomes a property, except the system columns
        // (wkb, time, end-time), the original geometry column, the per-vertex
        // array columns, and the geometry-component coordinate names — all
        // excluded via the SHARED `crate::input` predicates so every input
        // adaptor (file/PostGIS/DuckDB) excludes the identical set and tiles
        // don't carry coordinates twice. Columns whose type we can't map
        // (PostGIS `geometry`, unmapped arrays, …) decode to None at read time
        // and are silently dropped per-row.
        let mut props = Vec::new();
        for (idx, col) in columns.iter().enumerate() {
            let name = col.name();
            if idx == wkb_idx
                || idx == time.idx
                || end_time.as_ref().map(|e| e.idx) == Some(idx)
                || name == geom_column
                || crate::input::is_coordinate_column_name(name)
                || crate::input::is_vertex_metadata_column(name)
            {
                continue;
            }
            props.push(PropCol {
                idx,
                name: name.to_string(),
                ty: col.type_().clone(),
            });
        }

        Ok(RowSchema {
            wkb_idx,
            time,
            end_time,
            vertex_timestamps,
            vertex_values,
            vertex_value_matrix,
            props,
            time_format,
        })
    }

    fn parse_row(
        &self,
        row: &Row,
        time_strictness: InputStrictness,
        geometry_strictness: InputStrictness,
        row_no: usize,
        coercions: &mut VertexCoercions,
    ) -> Result<RowOutcome> {
        // Geometry (EWKB bytea -> GeoJSON + centroid lon/lat).
        let wkb: Option<Vec<u8>> = row.try_get(self.wkb_idx).ok();
        let parsed = wkb.as_deref().and_then(parse_wkb_geometry);
        let Some((geometry, lon, lat)) = parsed else {
            if geometry_strictness == InputStrictness::Strict {
                anyhow::bail!(
                    "row {row_no}: null or unparseable geometry (rerun without --strict-geometry to skip)"
                );
            }
            return Ok(RowOutcome::GeomSkip);
        };

        // Timestamp.
        let timestamp = match decode_time(row, &self.time, self.time_format, row_no)? {
            Some(ts) => ts,
            None => {
                if time_strictness == InputStrictness::Strict {
                    anyhow::bail!(
                        "row {row_no}: null/unparseable timestamp (rerun without --strict-times to coerce to epoch 0)"
                    );
                }
                // Warn mode mirrors the GeoParquet path: coerce to epoch 0.
                0
            }
        };
        let end_timestamp = match &self.end_time {
            Some(tc) => decode_time(row, tc, self.time_format, row_no)?,
            None => None,
        };

        // Per-vertex arrays (None when the column is absent or SQL NULL),
        // mirroring the GeoParquet reader's `vertex_*` columns.
        let vertex_timestamps = match &self.vertex_timestamps {
            Some(c) => decode_u64_ms_array(row, c, row_no, &mut coercions.timestamps)?,
            None => None,
        };
        let vertex_values = match &self.vertex_values {
            Some(c) => decode_f32_array(row, c, &mut coercions.values),
            None => None,
        };
        let vertex_value_matrix = match &self.vertex_value_matrix {
            Some(c) => decode_f32_array(row, c, &mut coercions.values),
            None => None,
        };

        // Properties.
        let mut properties = serde_json::Map::new();
        for p in &self.props {
            if let Some(value) = decode_property(row, p) {
                properties.insert(p.name.clone(), value);
            }
        }
        let shared_properties = if properties.is_empty() {
            None
        } else {
            Some(Arc::new(properties))
        };

        let feature = Feature {
            bbox: None,
            geometry: Some(geometry),
            id: None,
            properties: None,
            foreign_members: None,
        };
        Ok(RowOutcome::Feature(Box::new(ParsedFeature {
            geojson: feature,
            shared_properties,
            timestamp,
            end_timestamp,
            vertex_timestamps,
            vertex_values,
            vertex_value_matrix,
            lon,
            lat,
        })))
    }
}

/// Decode a time column value to Unix milliseconds, mirroring the GeoParquet
/// reader's per-type rules. Returns `Ok(None)` for SQL NULL / unmappable type.
fn decode_time(row: &Row, col: &TimeCol, time_format: TimeFormat, row_no: usize) -> Result<Option<u64>> {
    let ms: Option<i64> = match col.ty {
        Type::TIMESTAMPTZ => row
            .try_get::<_, Option<DateTime<Utc>>>(col.idx)
            .ok()
            .flatten()
            .map(|dt| dt.timestamp_millis()),
        Type::TIMESTAMP => row
            .try_get::<_, Option<NaiveDateTime>>(col.idx)
            .ok()
            .flatten()
            .map(|dt| dt.and_utc().timestamp_millis()),
        Type::DATE => row
            .try_get::<_, Option<NaiveDate>>(col.idx)
            .ok()
            .flatten()
            .and_then(|d| d.and_hms_opt(0, 0, 0))
            .map(|dt| dt.and_utc().timestamp_millis()),
        Type::INT8 => row
            .try_get::<_, Option<i64>>(col.idx)
            .ok()
            .flatten()
            .map(|v| apply_int_time_format(v, time_format, row_no))
            .transpose()?,
        Type::INT4 => row
            .try_get::<_, Option<i32>>(col.idx)
            .ok()
            .flatten()
            .map(|v| apply_int_time_format(v as i64, time_format, row_no))
            .transpose()?,
        Type::TEXT | Type::VARCHAR | Type::BPCHAR | Type::NAME => row
            .try_get::<_, Option<String>>(col.idx)
            .ok()
            .flatten()
            .and_then(|s| parse_iso8601(&s).ok()),
        _ => None,
    };
    match ms {
        Some(v) => {
            reject_negative_timestamp(row_no, v)?;
            Ok(Some(v as u64))
        }
        None => Ok(None),
    }
}

/// Decode a per-vertex timestamp array column to `Vec<u64>` ms-since-epoch,
/// mirroring the GeoParquet reader's `vertex_timestamps` rules: integer arrays
/// are RAW ms (the per-vertex column is never reinterpreted via `--time-format`,
/// matching `extract_vertex_timestamps_from_batch`), timestamp/date arrays
/// convert to UTC ms, a whole-cell SQL NULL yields `None`, a NULL element
/// becomes `0` (preserving per-vertex alignment), and any negative (pre-1970)
/// value is a hard error in both strictness modes (the shared stt-core guard).
fn decode_u64_ms_array(
    row: &Row,
    col: &TimeCol,
    row_no: usize,
    coerced: &mut usize,
) -> Result<Option<Vec<u64>>> {
    // Count each element that is NULL (or, for DATE, an unrepresentable value) —
    // it is silently defaulted to 0 below, so the summary can flag the gaps.
    let mut count_nulls = |nulls: usize| *coerced += nulls;
    let ms: Option<Vec<i64>> = match col.ty {
        Type::INT8_ARRAY => row
            .try_get::<_, Option<Vec<Option<i64>>>>(col.idx)
            .ok()
            .flatten()
            .map(|v| {
                count_nulls(v.iter().filter(|x| x.is_none()).count());
                v.into_iter().map(|x| x.unwrap_or(0)).collect()
            }),
        Type::INT4_ARRAY => row
            .try_get::<_, Option<Vec<Option<i32>>>>(col.idx)
            .ok()
            .flatten()
            .map(|v| {
                count_nulls(v.iter().filter(|x| x.is_none()).count());
                v.into_iter().map(|x| x.map(i64::from).unwrap_or(0)).collect()
            }),
        Type::TIMESTAMP_ARRAY => row
            .try_get::<_, Option<Vec<Option<NaiveDateTime>>>>(col.idx)
            .ok()
            .flatten()
            .map(|v| {
                count_nulls(v.iter().filter(|x| x.is_none()).count());
                v.into_iter()
                    .map(|x| x.map(|t| t.and_utc().timestamp_millis()).unwrap_or(0))
                    .collect()
            }),
        Type::TIMESTAMPTZ_ARRAY => row
            .try_get::<_, Option<Vec<Option<DateTime<Utc>>>>>(col.idx)
            .ok()
            .flatten()
            .map(|v| {
                count_nulls(v.iter().filter(|x| x.is_none()).count());
                v.into_iter()
                    .map(|x| x.map(|t| t.timestamp_millis()).unwrap_or(0))
                    .collect()
            }),
        Type::DATE_ARRAY => row
            .try_get::<_, Option<Vec<Option<NaiveDate>>>>(col.idx)
            .ok()
            .flatten()
            .map(|v| {
                count_nulls(
                    v.iter()
                        .filter(|x| {
                            x.and_then(|d| d.and_hms_opt(0, 0, 0)).is_none()
                        })
                        .count(),
                );
                v.into_iter()
                    .map(|x| {
                        x.and_then(|d| d.and_hms_opt(0, 0, 0))
                            .map(|t| t.and_utc().timestamp_millis())
                            .unwrap_or(0)
                    })
                    .collect()
            }),
        _ => None,
    };
    match ms {
        Some(v) => {
            for &x in &v {
                reject_negative_timestamp(row_no, x)?;
            }
            Ok(Some(v.into_iter().map(|x| x as u64).collect()))
        }
        None => Ok(None),
    }
}

/// Decode a per-vertex float array column to `Vec<f32>`, mirroring the
/// GeoParquet reader's `vertex_values` rules: `float8[]`/`float4[]` map to f32,
/// a whole-cell SQL NULL yields `None`, and a NULL element becomes `f32::NAN`
/// (preserving per-vertex alignment).
fn decode_f32_array(row: &Row, col: &TimeCol, coerced: &mut usize) -> Option<Vec<f32>> {
    match col.ty {
        Type::FLOAT8_ARRAY => row
            .try_get::<_, Option<Vec<Option<f64>>>>(col.idx)
            .ok()
            .flatten()
            .map(|v| {
                *coerced += v.iter().filter(|x| x.is_none()).count();
                v.into_iter()
                    .map(|x| x.map(|n| n as f32).unwrap_or(f32::NAN))
                    .collect()
            }),
        Type::FLOAT4_ARRAY => row
            .try_get::<_, Option<Vec<Option<f32>>>>(col.idx)
            .ok()
            .flatten()
            .map(|v| {
                *coerced += v.iter().filter(|x| x.is_none()).count();
                v.into_iter().map(|x| x.unwrap_or(f32::NAN)).collect()
            }),
        _ => None,
    }
}

/// Map one property column value to a JSON value. SQL NULL and unmappable PG
/// types (geometry, uuid, arrays, …) return `None` and are dropped.
///
/// This is a documented **superset** of `crate::input::extract_property_value`,
/// not an exact mirror: it shares the six core types (bool / Int32 / Int64 /
/// Float32 / Float64 / text) and the NaN→JSON-null rule, but additionally maps
/// `INT2`, `NUMERIC` (→ nearest-f64 number via [`decimal_string_to_json`] — the
/// file reader drops Parquet decimals), `TIMESTAMP`/`TIMESTAMPTZ`/`DATE`
/// (→ epoch-ms numbers) and `JSON`/`JSONB` — types a GeoParquet file can't
/// express the same way, so they only ever add properties a file build couldn't
/// carry, never diverge on the common type set. The PostGIS and DuckDB readers
/// stay in lockstep on that common set (NUMERIC/DECIMAL uses the one shared
/// conversion); `JSON`/`JSONB` parsed to *nested* JSON is PostGIS-only, because
/// duckdb-rs surfaces DuckDB's JSON logical type as plain `ValueRef::Text`
/// (arrives as a string property there — see db-input-adaptors.md).
fn decode_property(row: &Row, col: &PropCol) -> Option<serde_json::Value> {
    use serde_json::Value;
    match col.ty {
        Type::BOOL => row
            .try_get::<_, Option<bool>>(col.idx)
            .ok()
            .flatten()
            .map(Value::Bool),
        Type::INT2 => row
            .try_get::<_, Option<i16>>(col.idx)
            .ok()
            .flatten()
            .map(|v| Value::from(v as i64)),
        Type::INT4 => row
            .try_get::<_, Option<i32>>(col.idx)
            .ok()
            .flatten()
            .map(|v| Value::from(v as i64)),
        Type::INT8 => row
            .try_get::<_, Option<i64>>(col.idx)
            .ok()
            .flatten()
            .map(Value::from),
        Type::FLOAT4 => row
            .try_get::<_, Option<f32>>(col.idx)
            .ok()
            .flatten()
            .map(|v| json_number_or_null(v as f64)),
        Type::FLOAT8 => row
            .try_get::<_, Option<f64>>(col.idx)
            .ok()
            .flatten()
            .map(json_number_or_null),
        // Fixed-point NUMERIC → nearest f64, via the conversion shared with the
        // DuckDB reader's DECIMAL arm (identical value → identical number).
        Type::NUMERIC => row
            .try_get::<_, Option<rust_decimal::Decimal>>(col.idx)
            .ok()
            .flatten()
            .map(|d| decimal_string_to_json(&d.to_string())),
        Type::TEXT | Type::VARCHAR | Type::BPCHAR | Type::NAME => row
            .try_get::<_, Option<String>>(col.idx)
            .ok()
            .flatten()
            .map(Value::String),
        Type::TIMESTAMPTZ => row
            .try_get::<_, Option<DateTime<Utc>>>(col.idx)
            .ok()
            .flatten()
            .map(|dt| Value::from(dt.timestamp_millis())),
        Type::TIMESTAMP => row
            .try_get::<_, Option<NaiveDateTime>>(col.idx)
            .ok()
            .flatten()
            .map(|dt| Value::from(dt.and_utc().timestamp_millis())),
        Type::DATE => row
            .try_get::<_, Option<NaiveDate>>(col.idx)
            .ok()
            .flatten()
            .and_then(|d| d.and_hms_opt(0, 0, 0))
            .map(|dt| Value::from(dt.and_utc().timestamp_millis())),
        Type::JSON | Type::JSONB => row.try_get::<_, Option<Value>>(col.idx).ok().flatten(),
        // Unmappable (geometry, uuid, arrays, …) — drop silently.
        _ => None,
    }
}

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

    #[test]
    fn wraps_table_query_with_ewkb_projection() {
        let spec = QuerySpec {
            source: QuerySource::Table("public.trips".into()),
            geom_column: "geom".into(),
            where_clause: None,
            reproject_from_srid: None,
        };
        let q = spec.wrapped_query();
        assert!(q.contains("ST_AsEWKB(q.\"geom\")"), "{q}");
        assert!(q.contains(WKB_ALIAS), "{q}");
        assert!(q.contains("FROM ( SELECT * FROM public.trips ) AS q"), "{q}");
    }

    #[test]
    fn wraps_sql_with_reprojection_and_where() {
        let spec = QuerySpec {
            source: QuerySource::Sql("SELECT * FROM obs WHERE valid".into()),
            geom_column: "the_geom".into(),
            where_clause: Some("ts > now() - interval '1 day'".into()),
            reproject_from_srid: Some(3857),
        };
        let q = spec.wrapped_query();
        assert!(
            q.contains("ST_AsEWKB(ST_Transform(ST_SetSRID(q.\"the_geom\", 3857), 4326))"),
            "{q}"
        );
        assert!(q.contains("WHERE ts > now()"), "{q}");
    }

    #[test]
    fn quotes_identifiers_safely() {
        assert_eq!(QuerySpec::quote_ident("geom"), "\"geom\"");
        assert_eq!(QuerySpec::quote_ident("we\"ird"), "\"we\"\"ird\"");
    }

    #[test]
    fn tile_query_time_filter_honors_time_format() {
        let bbox = [-10.0, -5.0, 10.0, 5.0];
        // Default (Iso8601 → timestamp column): to_timestamp comparison.
        let iso =
            build_tile_query("obs", "geom", "ts", TimeFormat::Iso8601, bbox, 1000, 2000, None);
        assert!(iso.contains("to_timestamp(1000::double precision / 1000.0)"), "{iso}");
        assert!(iso.contains("to_timestamp(2000::double precision / 1000.0)"), "{iso}");

        // Integer-epoch-ms column: numeric comparison, no to_timestamp.
        let ms = build_tile_query("obs", "geom", "ts", TimeFormat::UnixMs, bbox, 1000, 2000, None);
        assert!(ms.contains("q.\"ts\" >= 1000 AND q.\"ts\" < 2000"), "{ms}");
        assert!(!ms.contains("to_timestamp"), "{ms}");

        // Seconds: sargable — the RAW column against ceil-divided second
        // bounds (t*1000 >= 1000 ⟺ t >= 1; t*1000 < 2000 ⟺ t < 2), never a
        // scaled-column expression that would defeat a b-tree index.
        let sec =
            build_tile_query("obs", "geom", "ts", TimeFormat::UnixSec, bbox, 1000, 2000, None);
        assert!(sec.contains("q.\"ts\" >= 1 AND q.\"ts\" < 2"), "{sec}");
        assert!(!sec.contains("* 1000"), "{sec}");

        // Non-multiple-of-1000 ms bounds round up on both ends (half-open).
        let odd = pg_time_window("t", TimeFormat::UnixSec, 1500, 2001);
        assert_eq!(odd, "t >= 2 AND t < 3");
    }

    #[test]
    fn tile_query_reprojects_when_source_srid_given() {
        let bbox = [-10.0, -5.0, 10.0, 5.0];
        let q = build_tile_query(
            "obs",
            "geom",
            "ts",
            TimeFormat::Iso8601,
            bbox,
            1000,
            2000,
            Some(3857),
        );
        // The transformed geometry feeds BOTH the EWKB projection and the bbox
        // test (the exact expression the --source-srid ingest uses), so the
        // filter runs in tile (4326) space — a strict superset.
        assert!(
            q.contains("ST_AsEWKB(ST_Transform(ST_SetSRID(q.\"geom\", 3857), 4326))"),
            "{q}"
        );
        assert!(
            q.contains(
                "ST_Transform(ST_SetSRID(q.\"geom\", 3857), 4326) && \
                 ST_MakeEnvelope(-10, -5, 10, 5, 4326)"
            ),
            "{q}"
        );
    }

    // ---- property_kind_for_pg: the schema-level Type → PropertyKind dispatch.
    //
    // The value-decode counterpart (`decode_property`) needs a live `Row`, so it
    // is covered end-to-end by the gated `postgres_decodes_*` parity tests. These
    // pin the pure classification for every arm — including NUMERIC and DATE,
    // which no cross-source parity fixture can carry (the file reader drops
    // Parquet decimals and can't express a DATE-as-property), so this is their
    // only non-live guard against a mis-classification silently dropping a column.

    #[test]
    fn property_kind_pg_classifies_numeric_types() {
        use crate::columnar::PropertyKind;
        for ty in [
            Type::INT2,
            Type::INT4,
            Type::INT8,
            Type::FLOAT4,
            Type::FLOAT8,
            Type::NUMERIC,
            Type::TIMESTAMP,
            Type::TIMESTAMPTZ,
            Type::DATE,
        ] {
            assert_eq!(
                property_kind_for_pg(&ty),
                Some(PropertyKind::Numeric),
                "{ty} must classify as Numeric"
            );
        }
    }

    #[test]
    fn property_kind_pg_classifies_categorical_types() {
        use crate::columnar::PropertyKind;
        for ty in [
            Type::BOOL,
            Type::TEXT,
            Type::VARCHAR,
            Type::BPCHAR,
            Type::NAME,
        ] {
            assert_eq!(
                property_kind_for_pg(&ty),
                Some(PropertyKind::Categorical),
                "{ty} must classify as Categorical"
            );
        }
    }

    #[test]
    fn property_kind_pg_drops_unmappable_and_nested() {
        // Arrays are vertex/time columns (handled by decode_u64_ms_array /
        // decode_f32_array), never scalar properties; geometry/uuid/bytea are
        // unmappable. JSON/JSONB are intentionally None at the SCHEMA layer even
        // though `decode_property` maps them at value time — nested JSON is left
        // to per-tile sniffing (see the `decode_property` doc-comment).
        for ty in [
            Type::JSON,
            Type::JSONB,
            Type::UUID,
            Type::BYTEA,
            Type::INT4_ARRAY,
            Type::INT8_ARRAY,
            Type::FLOAT8_ARRAY,
            Type::TIMESTAMP_ARRAY,
            Type::DATE_ARRAY,
        ] {
            assert_eq!(property_kind_for_pg(&ty), None, "{ty} must be dropped (None)");
        }
    }
}