nodedb_lite/engine/array/ops/
project.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! `ArrayOp::Project` handler for NodeDB-Lite.
4//!
5//! Scans all tiles (memtable + segments) for the named array, applies
6//! attribute projection via `nodedb_array::query::project::project_sparse`,
7//! and returns one row per live cell. The response mirrors the Slice arm:
8//! columns `["attrs", "valid_from_ms", "valid_until_ms"]`.
9
10use std::sync::Arc;
11
12use nodedb_array::query::project::{Projection, project_sparse};
13use nodedb_array::query::retention::decode_sparse_rows;
14use nodedb_array::tile::sparse_tile::RowKind;
15use nodedb_array::{SegmentReader, TilePayload};
16use nodedb_types::result::QueryResult;
17use nodedb_types::value::Value;
18
19use crate::engine::array::engine::ArrayEngineState;
20use crate::engine::array::ops::util::cell::cell_value_to_value;
21use crate::error::LiteError;
22use crate::runtime::now_millis_i64;
23use crate::storage::engine::StorageEngine;
24
25/// Execute `ArrayOp::Project` for the Lite engine.
26///
27/// Scans all segments and the memtable, projects to the requested attribute
28/// indices, and returns every live cell as a row. The attribute order in the
29/// response matches `attr_indices` — not the schema order.
30pub async fn project<S: StorageEngine>(
31    array_state: &Arc<tokio::sync::Mutex<ArrayEngineState>>,
32    storage: &Arc<S>,
33    name: &str,
34    attr_indices: &[u32],
35) -> Result<QueryResult, LiteError> {
36    let now_ms = now_millis_i64();
37
38    let (seg_ids, schema, schema_attr_count) = {
39        let state = array_state.lock().await;
40        let arr = state
41            .arrays
42            .get(name)
43            .ok_or_else(|| LiteError::BadRequest {
44                detail: format!("array '{name}' not found"),
45            })?;
46        let seg_ids: Vec<u64> = arr.manifest.segments.iter().map(|s| s.id).collect();
47        let attr_count = arr.schema.attrs.len();
48        (seg_ids, arr.schema.clone(), attr_count)
49    };
50
51    // Validate projection indices against schema attr count.
52    for &idx in attr_indices {
53        if idx as usize >= schema_attr_count {
54            return Err(LiteError::BadRequest {
55                detail: format!(
56                    "project attr index {idx} out of range (array '{name}' has {schema_attr_count} attrs)"
57                ),
58            });
59        }
60    }
61
62    let proj = Projection::new(attr_indices.iter().map(|&i| i as usize).collect());
63
64    let mut rows: Vec<Vec<Value>> = Vec::new();
65    let columns = vec![
66        "attrs".to_string(),
67        "valid_from_ms".to_string(),
68        "valid_until_ms".to_string(),
69    ];
70
71    // Segments.
72    for seg_id in &seg_ids {
73        let bytes = crate::engine::array::segments::load_segment(storage, name, *seg_id).await?;
74        let reader = SegmentReader::open(&bytes).map_err(|e| LiteError::Storage {
75            detail: format!("open segment {seg_id}: {e}"),
76        })?;
77        for idx in 0..reader.tile_count() {
78            let entry_tile_id = reader.tiles()[idx].tile_id;
79            if entry_tile_id.system_from_ms > now_ms {
80                continue;
81            }
82            let payload = reader.read_tile(idx).map_err(|e| LiteError::Storage {
83                detail: format!("read_tile seg {seg_id} idx {idx}: {e}"),
84            })?;
85            let sparse = match payload {
86                TilePayload::Sparse(s) => s,
87                TilePayload::Dense(_) => continue,
88            };
89            let projected = project_sparse(&sparse, &proj).map_err(|e| LiteError::Storage {
90                detail: format!("project_sparse: {e}"),
91            })?;
92            for row in decode_sparse_rows(&projected).map_err(|e| LiteError::Storage {
93                detail: format!("decode_sparse_rows: {e}"),
94            })? {
95                if row.kind != RowKind::Live {
96                    continue;
97                }
98                let p = match row.payload {
99                    Some(p) => p,
100                    None => continue,
101                };
102                let attrs_val =
103                    Value::Array(p.attrs.into_iter().map(cell_value_to_value).collect());
104                rows.push(vec![
105                    attrs_val,
106                    Value::Integer(p.valid_from_ms),
107                    Value::Integer(p.valid_until_ms),
108                ]);
109            }
110        }
111    }
112
113    // Memtable.
114    {
115        let mut state = array_state.lock().await;
116        let arr = state
117            .arrays
118            .get_mut(name)
119            .ok_or_else(|| LiteError::BadRequest {
120                detail: format!("array '{name}' not found"),
121            })?;
122        let mem_tiles = arr
123            .memtable
124            .drain_all_tiles_read_only(now_ms, &schema)
125            .map_err(|e| LiteError::Storage {
126                detail: format!("memtable drain: {e}"),
127            })?;
128        for (_tile_id, sparse) in &mem_tiles {
129            let projected = project_sparse(sparse, &proj).map_err(|e| LiteError::Storage {
130                detail: format!("project_sparse memtable: {e}"),
131            })?;
132            for row in decode_sparse_rows(&projected).map_err(|e| LiteError::Storage {
133                detail: format!("decode_sparse_rows memtable: {e}"),
134            })? {
135                if row.kind != RowKind::Live {
136                    continue;
137                }
138                let p = match row.payload {
139                    Some(p) => p,
140                    None => continue,
141                };
142                let attrs_val =
143                    Value::Array(p.attrs.into_iter().map(cell_value_to_value).collect());
144                rows.push(vec![
145                    attrs_val,
146                    Value::Integer(p.valid_from_ms),
147                    Value::Integer(p.valid_until_ms),
148                ]);
149            }
150        }
151    }
152
153    Ok(QueryResult {
154        columns,
155        rows,
156        rows_affected: 0,
157        command: None,
158    })
159}
nodedb_lite/engine/array/ops/project.rs

nodedb_lite/engine/array/ops/
project.rs
