Function binary_row_project

nodedb/src/data/executor/handlers/join/mod.rs:222–272

Apply projection to a binary msgpack row, keeping only requested columns. Projection names may be unqualified ("name") while keys are qualified ("users.name"). Returns a new msgpack map with only matching fields, using the unqualified name as the output key.

(
    row: &[u8],
    projection: &[nodedb_physical::physical_plan::JoinProjection],
)

/// ("users.name"). Returns a new msgpack map with only matching fields,
/// using the unqualified name as the output key.
pub(super) fn binary_row_project(
    row: &[u8],
    projection: &[nodedb_physical::physical_plan::JoinProjection],
) -> Vec<u8> {
    let Some((count, pos)) = msgpack_scan::map_header(row, 0) else {
        return row.to_vec();
    };

    // First pass: find matching entries.
    struct Entry {
        output_key: String,
        val_start: usize,
        val_end: usize,
    }
    let mut entries = Vec::with_capacity(projection.len());
    let mut scan_pos = pos;
    for _ in 0..count {
        let key = msgpack_scan::read_str(row, scan_pos);
        scan_pos = match msgpack_scan::skip_value(row, scan_pos) {
            Some(p) => p,
            None => break,
        };
        let val_start = scan_pos;
        scan_pos = match msgpack_scan::skip_value(row, scan_pos) {
            Some(p) => p,
            None => break,
        };
        if let Some(k) = &key {
            let short = k.rsplit('.').next().unwrap_or(k);
            if let Some(projected) = projection
                .iter()
                .find(|p| p.source == short || p.source == *k)
            {
                entries.push(Entry {
                    output_key: projected.output.clone(),
                    val_start,
                    val_end: scan_pos,
                });
            }
        }
    }

    // Build output map.
    let mut buf = Vec::with_capacity(row.len());
    write_map_header(&mut buf, entries.len());
    for e in &entries {
        write_str(&mut buf, &e.output_key);
        buf.extend_from_slice(&row[e.val_start..e.val_end]);
    }
    buf
}

#[cfg(test)]
mod tests {