nodedb-lite 0.1.0

// SPDX-License-Identifier: Apache-2.0

//! DDL lowerings: `CreateArray`, `DropArray`, `AlterArray`.

use nodedb_array::types::ArrayId;
use nodedb_physical::PhysicalTaskVisitor;
use nodedb_physical::physical_plan::ArrayOp;
use nodedb_sql::CreateArrayVisitArgs;
use nodedb_types::result::QueryResult;

use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::query::meta_ops;
use crate::query::physical_visitor::LiteDataPlaneVisitor;
use crate::query::visitor::adapter::LiteFut;
use crate::storage::engine::StorageEngine;

use super::schema::{LITE_TENANT, build_schema};

/// Lower `SqlPlan::CreateArray` → `ArrayOp::OpenArray`.
pub(crate) fn lower_create_array<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    args: CreateArrayVisitArgs<'_>,
) -> Result<LiteFut<'a>, LiteError> {
    let CreateArrayVisitArgs {
        name,
        dims,
        attrs,
        tile_extents,
        cell_order,
        tile_order,
        prefix_bits,
        audit_retain_ms,
        minimum_audit_retain_ms,
    } = args;
    let schema = build_schema(name, dims, attrs, tile_extents, cell_order, tile_order)?;
    let schema_bytes = zerompk::to_msgpack_vec(&schema).map_err(|e| LiteError::Serialization {
        detail: format!("encode array schema: {e}"),
    })?;
    let schema_hash = crate::engine::array::catalog::hash_schema(&schema)?;
    let aid = ArrayId::new(LITE_TENANT, name);
    // Retention is held as milliseconds since epoch arithmetic downstream
    // (`now_ms - audit_retain_ms`), so it is signed there. A value past
    // `i64::MAX` is refused rather than cast: wrapping it negative would put
    // the purge horizon in the future and delete every tile version the array
    // has, which is the opposite of what a retention setting asks for.
    let audit_retain_ms = match audit_retain_ms {
        Some(ms) => Some(i64::try_from(ms).map_err(|_| LiteError::BadRequest {
            detail: format!(
                "CREATE ARRAY {name}: audit_retain_ms {ms} exceeds the maximum \
                 supported retention of {} ms",
                i64::MAX
            ),
        })?),
        None => None,
    };
    let op = ArrayOp::OpenArray {
        array_id: aid,
        schema_msgpack: schema_bytes,
        schema_hash,
        prefix_bits,
        audit_retain_ms,
        minimum_audit_retain_ms,
    };
    let mut phys = LiteDataPlaneVisitor::new(engine);
    let fut = phys.array(&op)?;
    Ok(Box::pin(fut))
}

/// Lower `SqlPlan::DropArray` → `ArrayOp::DropArray`.
pub(crate) fn lower_drop_array<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    name: &str,
    if_exists: bool,
) -> Result<LiteFut<'a>, LiteError> {
    let name_owned = name.to_string();
    Ok(Box::pin(async move {
        let exists = engine
            .array_state
            .lock()
            .await
            .arrays
            .contains_key(&name_owned);
        if !exists {
            if if_exists {
                return Ok(QueryResult {
                    columns: vec![],
                    rows: vec![],
                    rows_affected: 0,
                    command: None,
                });
            }
            return Err(LiteError::BadRequest {
                detail: format!("DROP ARRAY: array '{name_owned}' not found"),
            });
        }
        let aid = ArrayId::new(LITE_TENANT, &name_owned);
        let op = ArrayOp::DropArray { array_id: aid };
        let mut phys = LiteDataPlaneVisitor::new(engine);
        phys.array(&op)?.await
    }))
}

/// Lower `SqlPlan::AlterArray` → `meta_ops::handle_alter_array`.
pub(crate) fn lower_alter_array<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    name: &str,
    audit_retain_ms: Option<Option<i64>>,
    minimum_audit_retain_ms: Option<u64>,
) -> Result<LiteFut<'a>, LiteError> {
    let name = name.to_string();
    let min_wrap: Option<Option<u64>> = minimum_audit_retain_ms.map(Some);
    Ok(Box::pin(async move {
        meta_ops::handle_alter_array(engine, &name, audit_retain_ms, min_wrap).await
    }))
}

#[cfg(test)]
mod tests {
    use nodedb_sql::types_array::ArrayCellOrderAst as Cell;
    use nodedb_sql::types_array::ArrayTileOrderAst as Tile;

    use super::super::testing::{attr1_ast, dim1_ast, make_engine};
    use super::*;

    #[tokio::test]
    async fn test_create_array() {
        let engine = make_engine().await;
        let fut = lower_create_array(
            &engine,
            CreateArrayVisitArgs {
                name: "arr1",
                dims: &dim1_ast(),
                attrs: &attr1_ast(),
                tile_extents: &[4],
                cell_order: Cell::Hilbert,
                tile_order: Tile::Hilbert,
                prefix_bits: 0,
                audit_retain_ms: None,
                minimum_audit_retain_ms: None,
            },
        )
        .expect("lower");
        let r = fut.await.expect("execute");
        assert_eq!(r.rows_affected, 1);
    }

    #[tokio::test]
    async fn test_drop_array() {
        let engine = make_engine().await;
        lower_create_array(
            &engine,
            CreateArrayVisitArgs {
                name: "arr_drop",
                dims: &dim1_ast(),
                attrs: &attr1_ast(),
                tile_extents: &[4],
                cell_order: Cell::Hilbert,
                tile_order: Tile::Hilbert,
                prefix_bits: 0,
                audit_retain_ms: None,
                minimum_audit_retain_ms: None,
            },
        )
        .unwrap()
        .await
        .unwrap();

        let fut = lower_drop_array(&engine, "arr_drop", false).expect("lower");
        let r = fut.await.expect("execute");
        assert_eq!(r.rows_affected, 1);
    }

    #[tokio::test]
    async fn test_drop_array_if_exists_missing() {
        let engine = make_engine().await;
        let fut = lower_drop_array(&engine, "nonexistent", true).expect("lower");
        let r = fut.await.expect("execute");
        assert_eq!(r.rows_affected, 0);
    }

    #[tokio::test]
    async fn test_alter_array() {
        let engine = make_engine().await;
        lower_create_array(
            &engine,
            CreateArrayVisitArgs {
                name: "arr_alt",
                dims: &dim1_ast(),
                attrs: &attr1_ast(),
                tile_extents: &[4],
                cell_order: Cell::Hilbert,
                tile_order: Tile::Hilbert,
                prefix_bits: 0,
                audit_retain_ms: None,
                minimum_audit_retain_ms: None,
            },
        )
        .unwrap()
        .await
        .unwrap();

        let fut = lower_alter_array(&engine, "arr_alt", Some(Some(86400000)), None).expect("lower");
        let r = fut.await.expect("execute");
        assert_eq!(r.rows_affected, 1);
    }
}