use crate::owner::OwnerState;
use crate::r_api::{
c_string, RApi, Sexp, CE_UTF8, INTSXP, LGLSXP, RAWSXP, REALSXP, STRSXP, VECSXP,
};
use crate::work::{
DataFrameColumn, DataFrameValue, DataFrameWire, DecodeDataFrameWork, EncodeDataFrameWork,
NaType, WorkResult,
};
use rustler::{Encoder, Env, Term};
use std::collections::HashSet;
use std::sync::atomic::Ordering;
const INHERITS_DATAFRAME_SOURCE: &str = r#"inherits(.rx_df, "data.frame")"#;
const NROW_SOURCE: &str = r#"nrow(.rx_df)"#;
const DEFAULT_ROW_NAMES_SOURCE: &str = r#".row_names_info(.rx_df, 1L) <= 0L"#;
pub fn parse_wire(term: Term) -> Result<DataFrameWire, rustler::Error> {
let kind: String = map_get_term(term, "kind")?.decode()?;
if kind != "data_frame" {
return Err(rustler::Error::BadArg);
}
let names_terms: Vec<Term> = map_get_term(term, "names")?.decode()?;
let n_rows: usize = map_get_term(term, "n_rows")?.decode()?;
let column_terms: Vec<Term> = map_get_term(term, "columns")?.decode()?;
if names_terms.len() != column_terms.len() || (names_terms.is_empty() && n_rows > 0) {
return Err(rustler::Error::BadArg);
}
let mut names = Vec::with_capacity(names_terms.len());
let mut seen = HashSet::with_capacity(names_terms.len());
for name_term in names_terms {
let name = binary_bytes(name_term)?;
if name.is_empty() || !seen.insert(name.clone()) {
return Err(rustler::Error::BadArg);
}
names.push(name);
}
let mut columns = Vec::with_capacity(column_terms.len());
for (index, column_term) in column_terms.into_iter().enumerate() {
let name = binary_bytes(map_get_term(column_term, "name")?)?;
if name != names[index] {
return Err(rustler::Error::BadArg);
}
let column_type = parse_type_name(&map_get_term(column_term, "type")?.decode::<String>()?)?;
let value_terms: Vec<Term> = map_get_term(column_term, "values")?.decode()?;
if value_terms.len() != n_rows {
return Err(rustler::Error::BadArg);
}
let mut values = Vec::with_capacity(value_terms.len());
for value_term in value_terms {
values.push(parse_value(value_term, column_type)?);
}
columns.push(DataFrameColumn {
name,
column_type,
values,
});
}
Ok(DataFrameWire {
names,
n_rows,
columns,
})
}
pub fn parse_max_rows(opts: Term) -> Result<Option<usize>, rustler::Error> {
let entries = opts.decode::<Vec<Term>>()?;
let mut max_rows = None;
for entry in entries {
let (key, value): (Term, Term) = entry.decode()?;
if key.atom_to_string().ok().as_deref() == Some("max_rows") {
let rows = value.decode::<usize>()?;
if rows == 0 {
return Err(rustler::Error::BadArg);
}
max_rows = Some(rows);
}
}
Ok(max_rows)
}
pub fn encode_wire<'a>(env: Env<'a>, wire: &DataFrameWire) -> Term<'a> {
let names = wire
.names
.iter()
.map(|name| crate::terms::binary(env, name))
.collect::<Vec<_>>();
let columns = wire
.columns
.iter()
.map(|column| encode_column(env, column))
.collect::<Vec<_>>();
map_from_binary_keys(
env,
&[
(b"kind".as_slice(), "data_frame".encode(env)),
(b"names".as_slice(), names.encode(env)),
(b"n_rows".as_slice(), wire.n_rows.encode(env)),
(b"columns".as_slice(), columns.encode(env)),
],
)
}
pub fn encode_tagged_error<'a>(env: Env<'a>, tag: &'static str, message: &str) -> Term<'a> {
match tag {
"not_dataframe" => (
crate::atoms::error(),
(crate::atoms::not_dataframe(), message),
)
.encode(env),
"unsupported_dataframe_column" => {
let Some((name, reason)) = split_column_message(message) else {
return (crate::atoms::error(), message).encode(env);
};
(
crate::atoms::error(),
(
crate::atoms::unsupported_dataframe_column(),
crate::terms::binary(env, name.as_bytes()),
reason_term(env, reason),
),
)
.encode(env)
}
"unsupported_dataframe_names" => (
crate::atoms::error(),
(
crate::atoms::unsupported_dataframe_names(),
reason_term(env, message),
),
)
.encode(env),
"unsupported_dataframe_row_names" => (
crate::atoms::error(),
(
crate::atoms::unsupported_dataframe_row_names(),
reason_term(env, message),
),
)
.encode(env),
"unsupported_dataframe_shape" => (
crate::atoms::error(),
(
crate::atoms::unsupported_dataframe_shape(),
reason_term(env, message),
),
)
.encode(env),
"invalid_dataframe" => (
crate::atoms::error(),
(crate::atoms::invalid_dataframe(), reason_term(env, message)),
)
.encode(env),
"dataframe_too_large" => {
let mut parts = message.split(':').map(str::trim);
let rows = parts.next().and_then(|part| part.parse::<usize>().ok());
let max_rows = parts.next().and_then(|part| part.parse::<usize>().ok());
match (rows, max_rows) {
(Some(rows), Some(max_rows)) => (
crate::atoms::error(),
(crate::atoms::dataframe_too_large(), rows, max_rows),
)
.encode(env),
_ => (crate::atoms::error(), message).encode(env),
}
}
_ => (crate::atoms::error(), message).encode(env),
}
}
pub fn do_encode_data_frame(state: &mut OwnerState, work: &EncodeDataFrameWork) -> WorkResult {
if let Some(failure) = state.terminal_init_failure() {
return WorkResult::NativeInitFailed(failure);
}
if !state.is_initialized() {
return WorkResult::Error("embedded R runtime is not initialized".to_owned());
}
let Some(api) = state.r.as_ref() else {
return WorkResult::Error("embedded R API is not initialized".to_owned());
};
unsafe { do_encode_data_frame_inner(api, &work.wire) }
}
pub fn do_decode_data_frame(state: &mut OwnerState, work: &DecodeDataFrameWork) -> WorkResult {
if let Some(failure) = state.terminal_init_failure() {
return WorkResult::NativeInitFailed(failure);
}
if !state.is_initialized() {
return WorkResult::Error("embedded R runtime is not initialized".to_owned());
}
let Some(api) = state.r.as_ref() else {
return WorkResult::Error("embedded R API is not initialized".to_owned());
};
let (sexp, resource_kind) = {
let state = match work.resource.state.lock() {
Ok(state) => state,
Err(_error) => {
return WorkResult::Error("native R object resource mutex was poisoned".to_owned())
}
};
if work.resource.release_enqueued.load(Ordering::SeqCst) {
return WorkResult::Error(
"native R object resource has already been released".to_owned(),
);
}
match state.sexp {
Some(sexp) => (sexp, state.kind),
None => {
return WorkResult::Error(
"native R object resource has already been released".to_owned(),
)
}
}
};
unsafe { do_decode_data_frame_inner(api, sexp, resource_kind, work.max_rows) }
}
fn map_get_term<'a>(term: Term<'a>, key: &str) -> Result<Term<'a>, rustler::Error> {
let key_term = crate::terms::binary(term.get_env(), key.as_bytes());
term.map_get(key_term)
}
fn binary_bytes(term: Term) -> Result<Vec<u8>, rustler::Error> {
term.decode::<rustler::Binary>()
.map(|binary| binary.as_slice().to_vec())
}
fn parse_type_name(name: &str) -> Result<NaType, rustler::Error> {
match name {
"logical" => Ok(NaType::Logical),
"integer" => Ok(NaType::Integer),
"double" => Ok(NaType::Double),
"character" => Ok(NaType::Character),
_ => Err(rustler::Error::BadArg),
}
}
fn parse_value(term: Term, column_type: NaType) -> Result<DataFrameValue, rustler::Error> {
if let Ok(kind) = map_get_term(term, "kind").and_then(|term| term.decode::<String>()) {
if kind != "na" {
return Err(rustler::Error::BadArg);
}
let na_type = parse_type_name(&map_get_term(term, "type")?.decode::<String>()?)?;
if na_type != column_type {
return Err(rustler::Error::BadArg);
}
return Ok(DataFrameValue::Na(na_type));
}
match column_type {
NaType::Logical => term.decode::<bool>().map(DataFrameValue::Logical),
NaType::Integer => {
let value = term.decode::<i64>()?;
if value < -2_147_483_647 || value > i64::from(i32::MAX) {
return Err(rustler::Error::BadArg);
}
Ok(DataFrameValue::Integer(value as i32))
}
NaType::Double => {
let value = if term.is_float() {
term.decode::<f64>()?
} else {
term.decode::<i64>()? as f64
};
if !value.is_finite() {
return Err(rustler::Error::BadArg);
}
Ok(DataFrameValue::Double(value))
}
NaType::Character => binary_bytes(term).map(DataFrameValue::Character),
}
}
fn encode_column<'a>(env: Env<'a>, column: &DataFrameColumn) -> Term<'a> {
let values = column
.values
.iter()
.map(|value| encode_value(env, value))
.collect::<Vec<_>>();
map_from_binary_keys(
env,
&[
(b"name".as_slice(), crate::terms::binary(env, &column.name)),
(
b"type".as_slice(),
type_name(column.column_type).encode(env),
),
(b"values".as_slice(), values.encode(env)),
],
)
}
fn encode_value<'a>(env: Env<'a>, value: &DataFrameValue) -> Term<'a> {
match value {
DataFrameValue::Logical(value) => value.encode(env),
DataFrameValue::Integer(value) => value.encode(env),
DataFrameValue::Double(value) => value.encode(env),
DataFrameValue::Character(value) => crate::terms::binary(env, value),
DataFrameValue::Na(na_type) => map_from_binary_keys(
env,
&[
(b"kind".as_slice(), "na".encode(env)),
(b"type".as_slice(), type_name(*na_type).encode(env)),
],
),
}
}
fn map_from_binary_keys<'a>(env: Env<'a>, entries: &[(&[u8], Term<'a>)]) -> Term<'a> {
let keys = entries
.iter()
.map(|(key, _value)| crate::terms::binary(env, key))
.collect::<Vec<_>>();
let values = entries
.iter()
.map(|(_key, value)| *value)
.collect::<Vec<_>>();
crate::terms::map_from_terms(env, &keys, &values)
}
fn type_name(na_type: NaType) -> &'static str {
match na_type {
NaType::Logical => "logical",
NaType::Integer => "integer",
NaType::Double => "double",
NaType::Character => "character",
}
}
fn reason_term<'a>(env: Env<'a>, reason: &str) -> Term<'a> {
match reason {
"non_string" => crate::atoms::non_string().encode(env),
"empty" => crate::atoms::empty().encode(env),
"duplicate" => crate::atoms::duplicate().encode(env),
"custom" => crate::atoms::custom().encode(env),
"zero_column_nonzero_row" => crate::atoms::zero_column_nonzero_row().encode(env),
"factor" => crate::atoms::factor().encode(env),
"date" => crate::atoms::date().encode(env),
"posix" => crate::atoms::posix().encode(env),
"list" => crate::atoms::list().encode(env),
"matrix" => crate::atoms::matrix().encode(env),
"complex" => crate::atoms::complex().encode(env),
"raw" => crate::atoms::raw().encode(env),
"non_finite_double" => crate::atoms::non_finite_double().encode(env),
"unsupported_type" => crate::atoms::unsupported_type().encode(env),
"malformed_wire" => crate::atoms::malformed_wire().encode(env),
"column_key_mismatch" => crate::atoms::column_key_mismatch().encode(env),
"column_length_mismatch" => crate::atoms::column_length_mismatch().encode(env),
"duplicate_names" => crate::atoms::duplicate_names().encode(env),
"empty_name" => crate::atoms::empty_name().encode(env),
"mixed_column_type" => crate::atoms::mixed_column_type().encode(env),
_ => reason.encode(env),
}
}
fn split_column_message(message: &str) -> Option<(&str, &str)> {
message.split_once(": ")
}
unsafe fn do_encode_data_frame_inner(api: &RApi, wire: &DataFrameWire) -> WorkResult {
let mut protect_count = 0;
let df = (api.protect)((api.alloc_vector)(VECSXP, wire.columns.len() as isize));
protect_count += 1;
let names = (api.protect)((api.alloc_vector)(STRSXP, wire.names.len() as isize));
protect_count += 1;
for (index, column) in wire.columns.iter().enumerate() {
let column_sexp = match make_column(api, column, &mut protect_count) {
Ok(column_sexp) => column_sexp,
Err(result) => {
(api.unprotect)(protect_count);
return result;
}
};
(api.set_vector_elt)(df, index as isize, column_sexp);
let name = match bytes_to_c_string(&wire.names[index]) {
Ok(name) => name,
Err(result) => {
(api.unprotect)(protect_count);
return result;
}
};
let name_sexp = (api.protect)((api.mk_char_len_ce)(
name.as_ptr(),
wire.names[index].len() as i32,
CE_UTF8,
));
(api.set_string_elt)(names, index as isize, name_sexp);
(api.unprotect)(1);
}
let class_value = (api.protect)((api.alloc_vector)(STRSXP, 1));
protect_count += 1;
let class_name = c_string("data.frame").expect("literal contains no NUL bytes");
let class_name_sexp = (api.protect)((api.mk_char_len_ce)(class_name.as_ptr(), 10, CE_UTF8));
(api.set_string_elt)(class_value, 0, class_name_sexp);
(api.unprotect)(1);
let row_names = make_row_names(api, wire.n_rows, &mut protect_count);
(api.set_attrib)(df, *api.names_symbol, names);
(api.set_attrib)(df, *api.class_symbol, class_value);
let row_names_symbol = c_string("row.names").expect("literal contains no NUL bytes");
(api.set_attrib)(df, (api.install)(row_names_symbol.as_ptr()), row_names);
let resource =
crate::codec::preserved_resource(api, df, crate::resource::ResourceKind::Dataframe);
(api.unprotect)(protect_count);
WorkResult::Resource(resource)
}
unsafe fn do_decode_data_frame_inner(
api: &RApi,
sexp: Sexp,
resource_kind: crate::resource::ResourceKind,
max_rows: Option<usize>,
) -> WorkResult {
let rx_df = c_string(".rx_df").expect(".rx_df contains no NUL bytes");
let mut protect_count = 0;
let env = (api.protect)((api.new_env)(*api.global_env, 1, 29));
protect_count += 1;
(api.define_var)((api.install)(rx_df.as_ptr()), sexp, env);
if resource_kind != crate::resource::ResourceKind::Dataframe {
match eval_logical_in_env(api, env, INHERITS_DATAFRAME_SOURCE) {
Ok(true) => {}
Ok(false) => {
(api.unprotect)(protect_count);
return WorkResult::TaggedError(
"not_dataframe",
"native object is not a data.frame".to_owned(),
);
}
Err(result) => {
(api.unprotect)(protect_count);
return result;
}
}
}
let column_count = match usize::try_from((api.xlength)(sexp)) {
Ok(count) => count,
Err(_error) => {
(api.unprotect)(protect_count);
return WorkResult::TaggedError("invalid_dataframe", "malformed_wire".to_owned());
}
};
let n_rows = match eval_usize_in_env(api, env, NROW_SOURCE) {
Ok(rows) => rows,
Err(result) => {
(api.unprotect)(protect_count);
return result;
}
};
if let Some(max_rows) = max_rows {
if n_rows > max_rows {
(api.unprotect)(protect_count);
return WorkResult::TaggedError("dataframe_too_large", format!("{n_rows}: {max_rows}"));
}
}
if column_count == 0 && n_rows > 0 {
(api.unprotect)(protect_count);
return WorkResult::TaggedError(
"unsupported_dataframe_shape",
"zero_column_nonzero_row".to_owned(),
);
}
match eval_logical_in_env(api, env, DEFAULT_ROW_NAMES_SOURCE) {
Ok(true) => {}
Ok(false) => {
(api.unprotect)(protect_count);
return WorkResult::TaggedError("unsupported_dataframe_row_names", "custom".to_owned());
}
Err(result) => {
(api.unprotect)(protect_count);
return result;
}
}
let names_attr = (api.get_attrib)(sexp, *api.names_symbol);
let names = match copy_names(api, names_attr, column_count) {
Ok(names) => names,
Err(result) => {
(api.unprotect)(protect_count);
return result;
}
};
let mut columns = Vec::with_capacity(column_count);
for (index, name) in names.iter().enumerate() {
let column_sexp = (api.vector_elt)(sexp, index as isize);
let column_type = match column_type_from_r(api, column_sexp) {
Ok(column_type) => column_type,
Err(reason) => {
(api.unprotect)(protect_count);
return WorkResult::TaggedError(
"unsupported_dataframe_column",
format!("{}: {reason}", String::from_utf8_lossy(name)),
);
}
};
let values = match copy_column_values(api, column_sexp, column_type, name) {
Ok(values) => values,
Err(result) => {
(api.unprotect)(protect_count);
return result;
}
};
if values.len() != n_rows {
(api.unprotect)(protect_count);
return WorkResult::TaggedError(
"invalid_dataframe",
"column_length_mismatch".to_owned(),
);
}
columns.push(DataFrameColumn {
name: name.clone(),
column_type,
values,
});
}
(api.unprotect)(protect_count);
WorkResult::DataFrameWire(DataFrameWire {
names,
n_rows,
columns,
})
}
unsafe fn make_column(
api: &RApi,
column: &DataFrameColumn,
protect_count: &mut i32,
) -> Result<Sexp, WorkResult> {
let len = column.values.len() as isize;
match column.column_type {
NaType::Logical => {
let vector = (api.protect)((api.alloc_vector)(LGLSXP, len));
*protect_count += 1;
for (index, value) in column.values.iter().enumerate() {
let encoded = match value {
DataFrameValue::Logical(value) => i32::from(*value),
DataFrameValue::Na(NaType::Logical) => *api.na_int,
_ => {
return Err(WorkResult::TaggedError(
"invalid_dataframe",
"mixed_column_type".to_owned(),
))
}
};
(api.set_logical_elt)(vector, index as isize, encoded);
}
Ok(vector)
}
NaType::Integer => {
let vector = (api.protect)((api.alloc_vector)(INTSXP, len));
*protect_count += 1;
for (index, value) in column.values.iter().enumerate() {
let encoded = match value {
DataFrameValue::Integer(value) => *value,
DataFrameValue::Na(NaType::Integer) => *api.na_int,
_ => {
return Err(WorkResult::TaggedError(
"invalid_dataframe",
"mixed_column_type".to_owned(),
))
}
};
(api.set_integer_elt)(vector, index as isize, encoded);
}
Ok(vector)
}
NaType::Double => {
let vector = (api.protect)((api.alloc_vector)(REALSXP, len));
*protect_count += 1;
for (index, value) in column.values.iter().enumerate() {
let encoded = match value {
DataFrameValue::Double(value) => *value,
DataFrameValue::Na(NaType::Double) => *api.na_real,
_ => {
return Err(WorkResult::TaggedError(
"invalid_dataframe",
"mixed_column_type".to_owned(),
))
}
};
(api.set_real_elt)(vector, index as isize, encoded);
}
Ok(vector)
}
NaType::Character => {
let vector = (api.protect)((api.alloc_vector)(STRSXP, len));
*protect_count += 1;
for (index, value) in column.values.iter().enumerate() {
match value {
DataFrameValue::Character(bytes) => {
let text = bytes_to_c_string(bytes)?;
let chars = (api.protect)((api.mk_char_len_ce)(
text.as_ptr(),
bytes.len() as i32,
CE_UTF8,
));
(api.set_string_elt)(vector, index as isize, chars);
(api.unprotect)(1);
}
DataFrameValue::Na(NaType::Character) => {
(api.set_string_elt)(vector, index as isize, *api.na_string);
}
_ => {
return Err(WorkResult::TaggedError(
"invalid_dataframe",
"mixed_column_type".to_owned(),
))
}
}
}
Ok(vector)
}
}
}
unsafe fn make_row_names(api: &RApi, n_rows: usize, protect_count: &mut i32) -> Sexp {
if n_rows == 0 {
let row_names = (api.protect)((api.alloc_vector)(INTSXP, 0));
*protect_count += 1;
return row_names;
}
let row_names = (api.protect)((api.alloc_vector)(INTSXP, 2));
*protect_count += 1;
(api.set_integer_elt)(row_names, 0, *api.na_int);
(api.set_integer_elt)(row_names, 1, -(n_rows as i32));
row_names
}
unsafe fn copy_names(api: &RApi, names: Sexp, count: usize) -> Result<Vec<Vec<u8>>, WorkResult> {
if count == 0 {
return Ok(Vec::new());
}
if names == *api.nil_value
|| (api.type_of)(names) != STRSXP
|| (api.xlength)(names) != count as isize
{
return Err(WorkResult::TaggedError(
"unsupported_dataframe_names",
"non_string".to_owned(),
));
}
let mut out = Vec::with_capacity(count);
let mut seen = HashSet::with_capacity(count);
for index in 0..count {
let name = (api.string_elt)(names, index as isize);
if name == *api.na_string {
return Err(WorkResult::TaggedError(
"unsupported_dataframe_names",
"non_string".to_owned(),
));
}
let bytes = copy_r_string(api, name);
if bytes.is_empty() {
return Err(WorkResult::TaggedError(
"unsupported_dataframe_names",
"empty".to_owned(),
));
}
if !seen.insert(bytes.clone()) {
return Err(WorkResult::TaggedError(
"unsupported_dataframe_names",
"duplicate".to_owned(),
));
}
out.push(bytes);
}
Ok(out)
}
unsafe fn column_type_from_r(api: &RApi, column: Sexp) -> Result<NaType, &'static str> {
let dim = (api.get_attrib)(column, *api.dim_symbol);
if dim != *api.nil_value && (api.xlength)(dim) > 0 {
return Err("matrix");
}
let class_attr = (api.get_attrib)(column, *api.class_symbol);
if class_attr != *api.nil_value && (api.xlength)(class_attr) > 0 {
if inherits(api, column, "factor") {
return Err("factor");
}
if inherits(api, column, "Date") {
return Err("date");
}
if inherits(api, column, "POSIXt") {
return Err("posix");
}
}
if (api.type_of)(column) == VECSXP {
return Err("list");
}
if class_attr != *api.nil_value && (api.xlength)(class_attr) > 0 {
return Err("unsupported_type");
}
match (api.type_of)(column) {
LGLSXP => Ok(NaType::Logical),
INTSXP => Ok(NaType::Integer),
REALSXP => Ok(NaType::Double),
STRSXP => Ok(NaType::Character),
15 => Err("complex"),
RAWSXP => Err("raw"),
_ => Err("unsupported_type"),
}
}
unsafe fn copy_column_values(
api: &RApi,
column: Sexp,
column_type: NaType,
name: &[u8],
) -> Result<Vec<DataFrameValue>, WorkResult> {
let len = usize::try_from((api.xlength)(column)).map_err(|_error| {
WorkResult::TaggedError("invalid_dataframe", "column_length_mismatch".to_owned())
})?;
let mut values = Vec::with_capacity(len);
for index in 0..len {
values.push(match column_type {
NaType::Logical => {
let value = (api.logical_elt)(column, index as isize);
if value == *api.na_int {
DataFrameValue::Na(NaType::Logical)
} else {
DataFrameValue::Logical(value == 1)
}
}
NaType::Integer => {
let value = (api.integer_elt)(column, index as isize);
if value == *api.na_int {
DataFrameValue::Na(NaType::Integer)
} else {
DataFrameValue::Integer(value)
}
}
NaType::Double => {
let value = (api.real_elt)(column, index as isize);
if value.is_nan() && value.to_bits() == (*api.na_real).to_bits() {
DataFrameValue::Na(NaType::Double)
} else if !value.is_finite() {
return Err(WorkResult::TaggedError(
"unsupported_dataframe_column",
format!("{}: non_finite_double", String::from_utf8_lossy(name)),
));
} else {
DataFrameValue::Double(value)
}
}
NaType::Character => {
let string = (api.string_elt)(column, index as isize);
if string == *api.na_string {
DataFrameValue::Na(NaType::Character)
} else {
DataFrameValue::Character(copy_r_string(api, string))
}
}
});
}
Ok(values)
}
unsafe fn eval_logical_in_env(api: &RApi, env: Sexp, source: &str) -> Result<bool, WorkResult> {
let value = eval_source_in_env(api, env, source)?;
if (api.type_of)(value) != LGLSXP || (api.xlength)(value) < 1 {
return Err(WorkResult::Error(
"R expression did not return a logical scalar".to_owned(),
));
}
Ok((api.logical_elt)(value, 0) == 1)
}
unsafe fn eval_usize_in_env(api: &RApi, env: Sexp, source: &str) -> Result<usize, WorkResult> {
let value = eval_source_in_env(api, env, source)?;
if (api.xlength)(value) < 1 {
return Err(WorkResult::Error(
"R expression did not return a scalar integer".to_owned(),
));
}
match (api.type_of)(value) {
INTSXP => usize::try_from((api.integer_elt)(value, 0)).map_err(|_error| {
WorkResult::Error("R expression returned a negative integer".to_owned())
}),
REALSXP => {
let value = (api.real_elt)(value, 0);
if !value.is_finite() || value < 0.0 || value.fract() != 0.0 {
Err(WorkResult::Error(
"R expression returned an invalid integer".to_owned(),
))
} else {
Ok(value as usize)
}
}
_ => Err(WorkResult::Error(
"R expression did not return an integer".to_owned(),
)),
}
}
unsafe fn eval_source_in_env(api: &RApi, env: Sexp, source: &str) -> Result<Sexp, WorkResult> {
let source = c_string(source)
.map_err(|error| WorkResult::Error(format!("R source contains NUL byte: {error}")))?;
let mut protect_count = 0;
let source_sexp = (api.protect)((api.mk_string)(source.as_ptr()));
protect_count += 1;
let mut parse_status = crate::r_api::ParseStatus::Null;
let exprs = (api.protect)((api.parse_vector)(
source_sexp,
-1,
&mut parse_status,
*api.nil_value,
));
protect_count += 1;
if parse_status != crate::r_api::ParseStatus::Ok
&& parse_status != crate::r_api::ParseStatus::Null
{
(api.unprotect)(protect_count);
return Err(WorkResult::Error("R parse failed".to_owned()));
}
let mut result = *api.nil_value;
for index in 0..(api.xlength)(exprs) {
let mut error_occurred = 0;
result = (api.try_eval)((api.vector_elt)(exprs, index), env, &mut error_occurred);
if error_occurred != 0 {
(api.unprotect)(protect_count);
return Err(WorkResult::Error("R evaluation failed".to_owned()));
}
}
(api.unprotect)(protect_count);
Ok(result)
}
unsafe fn inherits(api: &RApi, value: Sexp, class_name: &str) -> bool {
let Ok(rx_value) = c_string(".rx_value") else {
return false;
};
let Ok(source) = c_string(&format!("inherits(.rx_value, {class_name:?})")) else {
return false;
};
let mut protect_count = 0;
let env = (api.protect)((api.new_env)(*api.global_env, 1, 29));
protect_count += 1;
(api.define_var)((api.install)(rx_value.as_ptr()), value, env);
let result = eval_logical_in_env(api, env, source.to_str().unwrap_or(""));
(api.unprotect)(protect_count);
result.unwrap_or(false)
}
unsafe fn copy_r_string(api: &RApi, string: Sexp) -> Vec<u8> {
let len = (api.length)(string);
if len <= 0 {
return Vec::new();
}
let chars = (api.r_char)(string);
if chars.is_null() {
return Vec::new();
}
std::slice::from_raw_parts(chars.cast::<u8>(), len as usize).to_vec()
}
fn bytes_to_c_string(bytes: &[u8]) -> Result<std::ffi::CString, WorkResult> {
std::ffi::CString::new(bytes)
.map_err(|_error| WorkResult::TaggedError("invalid_dataframe", "malformed_wire".to_owned()))
}