Method op_to_data_plane_plan

nodedb/src/control/array_sync/inbound_propose.rs:213–265

Convert a decoded `ArrayOp` (from sync) into a `PhysicalPlan::Array` variant.

(
        &self,
        op: &ArrayOp,
    )

    /// Convert a decoded `ArrayOp` (from sync) into a `PhysicalPlan::Array` variant.
    pub(super) fn op_to_data_plane_plan(
        &self,
        op: &ArrayOp,
    ) -> Result<crate::bridge::envelope::PhysicalPlan, Option<ArrayRejectMsg>> {
        use nodedb_array::sync::op::ArrayOpKind;
        use nodedb_physical::physical_plan::ArrayOp as DataArrayOp;

        let array_id = nodedb_array::types::ArrayId::new(self.tenant_id(), &op.header.array);

        let data_op = match op.kind {
            ArrayOpKind::Put => {
                let cells = vec![crate::engine::array::wal::ArrayPutCell {
                    coord: op.coord.clone(),
                    attrs: op.attrs.clone().unwrap_or_default(),
                    surrogate: nodedb_types::Surrogate::ZERO,
                    system_from_ms: op.header.system_from_ms,
                    valid_from_ms: op.header.valid_from_ms,
                    valid_until_ms: op.header.valid_until_ms,
                }];
                let cells_msgpack = zerompk::to_msgpack_vec(&cells).map_err(|e| {
                    Some(build_reject(
                        &op.header.array,
                        op.header.hlc,
                        ArrayRejectReason::ShapeInvalid,
                        format!("cells encode: {e}"),
                    ))
                })?;
                DataArrayOp::Put {
                    array_id,
                    cells_msgpack,
                    wal_lsn: 0,
                }
            }
            ArrayOpKind::Delete | ArrayOpKind::Erase => {
                let coords = vec![op.coord.clone()];
                let coords_msgpack = zerompk::to_msgpack_vec(&coords).map_err(|e| {
                    Some(build_reject(
                        &op.header.array,
                        op.header.hlc,
                        ArrayRejectReason::ShapeInvalid,
                        format!("coords encode: {e}"),
                    ))
                })?;
                DataArrayOp::Delete {
                    array_id,
                    coords_msgpack,
                    wal_lsn: 0,
                }
            }
        };

        Ok(crate::bridge::envelope::PhysicalPlan::Array(data_op))
    }
}