nodedb-lite 0.1.0

// SPDX-License-Identifier: Apache-2.0
//! SQL-visitor lowering for timeseries SqlPlan variants:
//! TimeseriesScan, TimeseriesIngest.

use nodedb_physical::PhysicalTaskVisitor;
use nodedb_physical::physical_plan::TimeseriesOp;
use nodedb_sql::TimeseriesScanVisitArgs;
use nodedb_sql::temporal::TemporalScope;
use nodedb_sql::types::filter::Filter;
use nodedb_sql::types::query::AggregateExpr;
use nodedb_sql::types_expr::SqlExpr;
use nodedb_sql::types_expr::SqlValue;

use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::query::filter_convert::sql_filters_to_metadata;
use crate::query::physical_visitor::LiteDataPlaneVisitor;
use crate::query::visitor::scan_post::sort_rows;
use crate::storage::engine::StorageEngine;

use super::adapter::LiteFut;
use super::having_eval::{apply_having_result, make_agg_alias_map};
use super::projection::projection_names;

fn encode_filters(filters: &[Filter]) -> Result<Vec<u8>, LiteError> {
    if filters.is_empty() {
        return Ok(Vec::new());
    }
    // Complex QExpr predicates are evaluated post-scan; only primitive conditions
    // are pushed to the physical visitor via serialized MetadataFilter.
    match sql_filters_to_metadata(filters, &[])?.meta {
        None => Ok(Vec::new()),
        Some(mf) => zerompk::to_msgpack_vec(&mf).map_err(|e| LiteError::Serialization {
            detail: format!("encode timeseries filters: {e}"),
        }),
    }
}

/// Convert SQL `AggregateExpr` list to `(op, field)` pairs expected by `TimeseriesOp::Scan`.
///
/// The grouped scan reads one column per aggregate. A computed argument has
/// no column to read, so it is refused. `DISTINCT` selects the deduplicating
/// accumulator name.
fn convert_aggregates(aggregates: &[AggregateExpr]) -> Result<Vec<(String, String)>, LiteError> {
    aggregates
        .iter()
        .map(|agg| {
            let field = match agg.args.first() {
                None | Some(SqlExpr::Wildcard) => "*".to_string(),
                Some(SqlExpr::Column { name, .. }) => name.clone(),
                Some(_) => {
                    return Err(LiteError::Unsupported {
                        detail: format!(
                            "aggregate '{}' has a computed argument the grouped timeseries \
                             scan cannot read in Lite",
                            agg.alias
                        ),
                    });
                }
            };
            Ok((
                nodedb_sql::planner::agg_naming::aggregate_function_name(agg),
                field,
            ))
        })
        .collect()
}

// ── TimeseriesScan ────────────────────────────────────────────────────────────

/// Lower `SqlPlan::TimeseriesScan` to `TimeseriesOp::Scan`.
pub(super) fn lower_timeseries_scan<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    args: TimeseriesScanVisitArgs<'_>,
) -> Result<LiteFut<'a>, LiteError> {
    let TimeseriesScanVisitArgs {
        collection,
        time_range,
        bucket_interval_ms,
        group_by,
        aggregates,
        filters,
        having,
        projection,
        gap_fill,
        limit,
        sort_keys,
        temporal,
        ..
    } = args;
    let filter_bytes = encode_filters(filters)?;
    let proj_cols = projection_names(projection, "timeseries scan")?;
    let agg_pairs = convert_aggregates(aggregates)?;

    let (system_time, valid_at_ms) = extract_temporal(temporal);

    // ORDER BY and HAVING must see the FULL result before it is truncated, so
    // an ordered or filtered query returns the first `limit` qualifying rows
    // — not an arbitrary `limit` rows that were then filtered or sorted among
    // themselves. So with sort keys or HAVING we scan unbounded (`limit == 0`
    // is the scan's "no cap" encoding), filter, sort, and truncate here.
    // Otherwise the limit is pushed into the scan.
    let post_processed = !sort_keys.is_empty() || !having.is_empty();
    let scan_limit = if post_processed { 0 } else { limit };

    let op = TimeseriesOp::Scan {
        // Lite holds a bare collection name; DatabaseId::DEFAULT keeps it unqualified.
        collection: nodedb_types::QualifiedCollection::new(
            nodedb_types::DatabaseId::DEFAULT,
            collection,
        ),
        time_range,
        projection: proj_cols,
        limit: scan_limit,
        filters: filter_bytes,
        bucket_interval_ms,
        group_by: group_by.to_vec(),
        aggregates: agg_pairs,
        gap_fill: gap_fill.to_string(),
        computed_columns: Vec::new(),
        rls_filters: Vec::new(),
        system_time,
        valid_at_ms,
        sort_keys: Vec::new(),
        // HAVING is applied below, over the scan's grouped rows.
        having: Vec::new(),
        tiering: None,
    };

    let mut phys = LiteDataPlaneVisitor::new(engine);
    let fut = phys.timeseries(&op)?;
    if !post_processed {
        return Ok(Box::pin(fut));
    }

    let sort_keys = sort_keys.to_vec();
    let having = having.to_vec();
    let agg_alias_map = make_agg_alias_map(aggregates);
    Ok(Box::pin(async move {
        let result = fut.await?;
        let mut result = apply_having_result(result, &having, &agg_alias_map);
        sort_rows(&mut result, &sort_keys)?;
        if limit > 0 {
            result.rows.truncate(limit);
        }
        Ok(result)
    }))
}

/// Extract the system-time scope and valid-time point from a `TemporalScope`.
///
/// Returns `(SystemTimeScope, Option<i64>)`. The caller passes the full
/// `SystemTimeScope` to `TimeseriesOp::Scan` so that `AllVersions` is
/// preserved faithfully; the physical adapter then uses `.is_all_versions()`
/// and `.as_of_ms()` to drive the engine, matching Origin's behaviour.
fn extract_temporal(scope: &TemporalScope) -> (nodedb_types::SystemTimeScope, Option<i64>) {
    use nodedb_sql::temporal::ValidTime;
    let sys = scope.system_time;
    let valid = match &scope.valid_time {
        ValidTime::At(ms) => Some(*ms),
        _ => None,
    };
    (sys, valid)
}

// ── TimeseriesIngest ──────────────────────────────────────────────────────────

/// Lower `SqlPlan::TimeseriesIngest` to `TimeseriesOp::Ingest`.
///
/// Rows are serialized to MessagePack in the `samples` format expected by the
/// Lite timeseries engine. Each row is a flat `HashMap<String, Value>` encoded
/// with zerompk; the payload field holds the concatenated msgpack bytes of a
/// `Vec<HashMap<String, Value>>`.
pub(super) fn lower_timeseries_ingest<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    permit: &'a crate::engine::fts::coordinator::TextMutationPermit,
    collection: &str,
    rows: &[Vec<(String, SqlValue)>],
) -> Result<LiteFut<'a>, LiteError> {
    use nodedb_types::value::Value;
    use std::collections::HashMap;

    let row_maps: Vec<HashMap<String, Value>> = rows
        .iter()
        .map(|row| {
            row.iter()
                .map(|(col, sv)| {
                    let v = crate::query::filter_convert::sql_value_to_value(sv)?;
                    Ok((col.clone(), v))
                })
                .collect::<Result<HashMap<String, Value>, LiteError>>()
        })
        .collect::<Result<Vec<_>, LiteError>>()?;

    let payload = zerompk::to_msgpack_vec(&row_maps).map_err(|e| LiteError::Serialization {
        detail: format!("encode timeseries ingest payload: {e}"),
    })?;

    let op = TimeseriesOp::Ingest {
        // Lite holds a bare collection name; DatabaseId::DEFAULT keeps it unqualified.
        collection: nodedb_types::QualifiedCollection::new(
            nodedb_types::DatabaseId::DEFAULT,
            collection,
        ),
        payload,
        format: "samples".to_string(),
        wal_lsn: None,
        surrogates: Vec::new(),
        provenance: None,
        // Lite's planner produces no RLS program and no RETURNING projection;
        // the adapter rejects either if one ever appears.
        // Lite has no RLS policy engine: no write policy applies.
        rls_write_check: nodedb_types::RlsWriteCheck::NoPolicyApplies,
        returning: None,
        rls_filters: Vec::new(),
    };

    let mut phys = LiteDataPlaneVisitor {
        engine,
        permit: Some(permit),
    };
    let fut = phys.timeseries(&op)?;
    Ok(Box::pin(fut))
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use crate::PagedbStorageMem;
    use crate::query::engine::{LiteQueryEngine, LiteQueryEngineParams};

    async fn make_engine() -> LiteQueryEngine<PagedbStorageMem> {
        use std::sync::Mutex;
        let storage = Arc::new(
            PagedbStorageMem::open_in_memory()
                .await
                .expect("in-memory pagedb"),
        );
        let crdt = Arc::new(Mutex::new(
            crate::engine::crdt::CrdtEngine::new(1).expect("crdt"),
        ));
        let governor = crate::query::engine::test_governor();
        let strict = Arc::new(crate::engine::strict::StrictEngine::new(Arc::clone(
            &storage,
        )));
        let columnar = Arc::new(crate::engine::columnar::ColumnarEngine::new(
            Arc::clone(&storage),
            crate::query::engine::test_scoped_memory(&governor, nodedb_mem::EngineId::Columnar),
        ));
        let htap = Arc::new(crate::engine::htap::HtapBridge::new());
        let timeseries = Arc::new(Mutex::new(
            crate::engine::timeseries::engine::TimeseriesEngine::new(),
        ));
        let vector_state = Arc::new(crate::engine::vector::VectorState::new(
            Arc::clone(&storage),
            100,
            crate::query::engine::test_scoped_memory(&governor, nodedb_mem::EngineId::Vector),
        ));
        let array_state = Arc::new(tokio::sync::Mutex::new(
            crate::engine::array::engine::ArrayEngineState::open(&storage)
                .await
                .expect("array"),
        ));
        let fts_state = Arc::new(crate::engine::fts::FtsState::new(Arc::clone(&governor)));
        let spatial = Arc::new(Mutex::new(
            crate::engine::spatial::SpatialIndexManager::new(
                crate::query::engine::test_scoped_memory(&governor, nodedb_mem::EngineId::Spatial),
            ),
        ));
        LiteQueryEngine::new(LiteQueryEngineParams {
            crdt,
            strict,
            columnar,
            htap,
            storage,
            timeseries,
            vector_state,
            array_state,
            fts_state,
            sparse_state: Arc::new(crate::engine::sparse_vector::SparseVectorState::new()),
            spatial,
            csr: Arc::new(Mutex::new(std::collections::HashMap::new())),
            governor,
            kv_local: crate::query::engine::test_kv_local(),
        })
    }

    #[tokio::test]
    async fn test_timeseries_scan_lower() {
        use nodedb_sql::temporal::TemporalScope;
        let engine = make_engine().await;
        let result = super::lower_timeseries_scan(
            &engine,
            nodedb_sql::TimeseriesScanVisitArgs {
                collection: "metrics",
                time_range: (0, i64::MAX),
                bucket_interval_ms: 0,
                group_by: &[],
                aggregates: &[],
                filters: &[],
                having: &[],
                projection: &[],
                gap_fill: "",
                limit: 100,
                sort_keys: &[],
                tiered: false,
                temporal: &TemporalScope::default(),
            },
        );
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_timeseries_ingest_lower() {
        use nodedb_sql::types_expr::SqlValue;
        let engine = make_engine().await;
        let rows = vec![vec![
            ("ts".to_string(), SqlValue::Int(1_700_000_000_000)),
            ("value".to_string(), SqlValue::Float(42.0)),
        ]];
        let guard = engine.fts_state.admit_mutation().await;
        let result = super::lower_timeseries_ingest(&engine, guard.permit(), "metrics", &rows);
        assert!(result.is_ok());
    }
}