nodedb-lite 0.1.0

// SPDX-License-Identifier: Apache-2.0
//! ArrayOp dispatch for the Lite physical visitor.

use std::sync::Arc;

use nodedb_array::query::slice::Slice;
use nodedb_array::types::cell_value::value::CellValue;
use nodedb_array::types::coord::value::CoordValue;
use nodedb_physical::physical_plan::ArrayOp;
use nodedb_types::result::QueryResult;
use nodedb_types::value::Value;

use crate::engine::array::engine::CellWrite;
use crate::engine::array::ops::util::cell::cell_value_to_value;
use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::runtime::now_millis_i64;
use crate::storage::engine::StorageEngine;

use super::{LitePhysicalFut, execute_surrogate_scan};

/// Local mirror of `nodedb::engine::array::wal::ArrayPutCell` for
/// deserializing `ArrayOp::Put.cells_msgpack` without a dependency on the
/// Origin binary crate. Field order and types must match the Origin
/// definition exactly because zerompk encodes structs as positional arrays.
#[derive(serde::Deserialize, zerompk::FromMessagePack)]
struct PutCellWire {
    coord: Vec<CoordValue>,
    attrs: Vec<CellValue>,
    // Ignored: `ArrayEngineState::put_cell` binds the surrogate.
    _surrogate: nodedb_types::Surrogate,
    system_from_ms: i64,
    valid_from_ms: i64,
    valid_until_ms: i64,
}

pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    op: &ArrayOp,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    match op {
        ArrayOp::OpenArray {
            array_id,
            schema_msgpack,
            audit_retain_ms,
            minimum_audit_retain_ms,
            ..
        } => {
            let name = array_id.name.clone();
            let schema_bytes = schema_msgpack.clone();
            let audit_retain_ms = *audit_retain_ms;
            let minimum_audit_retain_ms = *minimum_audit_retain_ms;
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                let schema =
                    zerompk::from_msgpack(&schema_bytes).map_err(|e| LiteError::Serialization {
                        detail: format!("decode ArraySchema: {e}"),
                    })?;
                let mut state = array_state.lock().await;
                state
                    .create_array(
                        &storage,
                        &name,
                        schema,
                        audit_retain_ms,
                        minimum_audit_retain_ms,
                    )
                    .await?;
                Ok(QueryResult {
                    columns: vec![],
                    rows: vec![],
                    rows_affected: 1,
                    command: None,
                })
            }))
        }

        ArrayOp::Put {
            array_id,
            cells_msgpack,
            ..
        } => {
            let name = array_id.name.clone();
            let cells_bytes = cells_msgpack.clone();
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                let cells: Vec<PutCellWire> =
                    zerompk::from_msgpack(&cells_bytes).map_err(|e| LiteError::Serialization {
                        detail: format!("decode Put cells: {e}"),
                    })?;
                let mut state = array_state.lock().await;
                let mut rows_affected: u64 = 0;
                for cell in cells {
                    state
                        .put_cell(
                            &storage,
                            &name,
                            CellWrite {
                                coord: cell.coord,
                                attrs: cell.attrs,
                                system_from_ms: cell.system_from_ms,
                                valid_from_ms: cell.valid_from_ms,
                                valid_until_ms: cell.valid_until_ms,
                            },
                        )
                        .await?;
                    rows_affected += 1;
                }
                Ok(QueryResult {
                    columns: vec![],
                    rows: vec![],
                    rows_affected,
                    command: Some("INSERT".into()),
                })
            }))
        }

        ArrayOp::Delete {
            array_id,
            coords_msgpack,
            ..
        } => {
            let name = array_id.name.clone();
            let coords_bytes = coords_msgpack.clone();
            let array_state = Arc::clone(&engine.array_state);
            Ok(Box::pin(async move {
                let coords: Vec<Vec<CoordValue>> =
                    zerompk::from_msgpack(&coords_bytes).map_err(|e| LiteError::Serialization {
                        detail: format!("decode Delete coords: {e}"),
                    })?;
                let now = now_millis_i64();
                let mut state = array_state.lock().await;
                let mut rows_affected: u64 = 0;
                for coord in coords {
                    state.delete_cell(&name, coord, now)?;
                    rows_affected += 1;
                }
                Ok(QueryResult {
                    columns: vec![],
                    rows: vec![],
                    rows_affected,
                    command: Some("DELETE".into()),
                })
            }))
        }

        ArrayOp::Slice {
            array_id,
            slice_msgpack,
            system_time,
            ..
        } => {
            use nodedb_types::SystemTimeScope;
            // Array is point-in-time; all-versions audit is not supported.
            if system_time.is_all_versions() {
                return Err(LiteError::Unsupported {
                    detail: "AS OF SYSTEM TIME NULL (all-versions) is not supported on \
                             the array engine in Lite"
                        .into(),
                });
            }
            let name = array_id.name.clone();
            let slice_bytes = slice_msgpack.clone();
            let system_as_of = match system_time {
                SystemTimeScope::AsOf(ms) => *ms,
                _ => i64::MAX,
            };
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                let slice: Slice =
                    zerompk::from_msgpack(&slice_bytes).map_err(|e| LiteError::Serialization {
                        detail: format!("decode Slice predicate: {e}"),
                    })?;
                let mut state = array_state.lock().await;
                let cells = state
                    .slice(&storage, &name, slice.dim_ranges, system_as_of)
                    .await?;
                let columns = vec![
                    "attrs".to_string(),
                    "valid_from_ms".to_string(),
                    "valid_until_ms".to_string(),
                ];
                let rows: Vec<Vec<Value>> = cells
                    .into_iter()
                    .map(|payload| {
                        let attrs_val = Value::Array(
                            payload.attrs.into_iter().map(cell_value_to_value).collect(),
                        );
                        vec![
                            attrs_val,
                            Value::Integer(payload.valid_from_ms),
                            Value::Integer(payload.valid_until_ms),
                        ]
                    })
                    .collect();
                Ok(QueryResult {
                    columns,
                    rows,
                    rows_affected: 0,
                    command: None,
                })
            }))
        }

        ArrayOp::Flush { array_id, .. } => {
            let name = array_id.name.clone();
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                let mut state = array_state.lock().await;
                state.flush(&storage, &name).await?;
                Ok(QueryResult {
                    columns: vec![],
                    rows: vec![],
                    rows_affected: 0,
                    command: None,
                })
            }))
        }

        ArrayOp::DropArray { array_id } => {
            let name = array_id.name.clone();
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                let mut state = array_state.lock().await;
                state.delete_array(&storage, &name).await?;
                Ok(QueryResult {
                    columns: vec![],
                    rows: vec![],
                    rows_affected: 1,
                    command: None,
                })
            }))
        }

        // Origin stages `DROP ARRAY` reversibly and compensates across cores
        // with `PurgeArrayDrop`. Lite's `DropArray` deletes the manifest and
        // catalog entry outright, so no tombstone exists to purge. Answering
        // `Ok` would report a compensation that did not happen.
        ArrayOp::PurgeArrayDrop { array_id } => Err(LiteError::BadRequest {
            detail: format!(
                "purge of a staged array drop is not supported in NodeDB-Lite \
                 (array '{}'): a Lite drop leaves no tombstone to purge",
                array_id.name
            ),
        }),

        ArrayOp::RekeyArray { array_id, .. } => Err(LiteError::Unsupported {
            detail: format!(
                "RekeyArray (MOVE TENANT) is unsupported on the single-node Lite engine (array '{}')",
                array_id.name
            ),
        }),

        ArrayOp::Project {
            array_id,
            attr_indices,
        } => {
            let name = array_id.name.clone();
            let indices = attr_indices.clone();
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                crate::engine::array::ops::project::project(&array_state, &storage, &name, &indices)
                    .await
            }))
        }

        ArrayOp::Aggregate {
            array_id,
            attr_idx,
            reducer,
            group_by_dim,
            ..
        } => {
            let name = array_id.name.clone();
            let attr_idx = *attr_idx;
            let reducer = *reducer;
            let group_by_dim = *group_by_dim;
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                crate::engine::array::ops::aggregate::aggregate(
                    &array_state,
                    &storage,
                    &name,
                    attr_idx,
                    reducer,
                    group_by_dim,
                )
                .await
            }))
        }

        ArrayOp::Elementwise {
            left, right, op, ..
        } => {
            let left_name = left.name.clone();
            let right_name = right.name.clone();
            let op = *op;
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                crate::engine::array::ops::elementwise::elementwise_op(
                    &array_state,
                    &storage,
                    &left_name,
                    &right_name,
                    op,
                )
                .await
            }))
        }

        ArrayOp::Compact {
            array_id,
            audit_retain_ms,
        } => {
            let name = array_id.name.clone();
            let retain_ms = *audit_retain_ms;
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                crate::engine::array::ops::compact::compact(
                    &array_state,
                    &storage,
                    &name,
                    retain_ms,
                )
                .await
            }))
        }

        ArrayOp::SurrogateBitmapScan {
            array_id,
            slice_msgpack,
        } => {
            let name = array_id.name.clone();
            let slice_bytes = slice_msgpack.clone();
            let array_state = Arc::clone(&engine.array_state);
            let storage = Arc::clone(&engine.storage);
            Ok(Box::pin(async move {
                let bitmap =
                    execute_surrogate_scan(&array_state, &storage, &name, &slice_bytes).await?;
                let mut bitmap_bytes = Vec::new();
                bitmap
                    .serialize_into(&mut bitmap_bytes)
                    .map_err(|e| LiteError::Serialization {
                        detail: format!("serialize surrogate bitmap: {e}"),
                    })?;
                Ok(QueryResult {
                    columns: vec!["bitmap".to_string()],
                    rows: vec![vec![nodedb_types::value::Value::Bytes(bitmap_bytes)]],
                    rows_affected: 0,
                    command: None,
                })
            }))
        }
    }
}