use std::cell::Cell;
use std::ffi::CString;
use std::marker::PhantomData;
use std::sync::Arc;
use crate::error::{Error, Result};
use crate::ffi::{ChsBlock, ChsFilter, ChsSchema, EXPORT_NONE, cstring};
use crate::json::quote_bare_denormals;
use crate::library::{Column, Library};
use crate::result::{
BatchDoc, BatchResult, DocFlags, FilterDoc, FilterResult, Format, RowDoc, RowResult,
batch_result_of, filter_result_of, row_result_of,
};
pub const NO_SETTINGS: &[(&str, &str)] = &[];
pub const NO_PARAMS: &[(&str, &str)] = &[];
pub const SETTING_NOW_EPOCH_NANOS: &str = "chtypes_now_epoch_nanos";
pub const SETTING_CLOCK_OFFSET_NANOS: &str = "chtypes_clock_offset_nanos";
pub const SETTING_MAX_CLOCK_SKEW_NANOS: &str = "chtypes_max_clock_skew_nanos";
pub const SETTING_DEFAULT_EVAL_MEMORY_BYTES: &str = "chtypes_default_eval_memory_bytes";
pub const SETTING_DEFAULT_EVAL_WALL_NANOS: &str = "chtypes_default_eval_wall_nanos";
pub struct Schema {
lib: Arc<Library>,
handle: *mut ChsSchema,
columns: Vec<Column>,
_not_sync: PhantomData<Cell<()>>,
}
unsafe impl Send for Schema {}
impl Schema {
pub(crate) fn new(lib: Arc<Library>, handle: *mut ChsSchema, columns: Vec<Column>) -> Schema {
Schema {
lib,
handle,
columns,
_not_sync: PhantomData,
}
}
pub fn columns(&self) -> &[Column] {
&self.columns
}
pub fn library(&self) -> &Arc<Library> {
&self.lib
}
pub fn set_engine<K: AsRef<str>, V: AsRef<str>>(
&mut self,
engine: &str,
order_by: &str,
merge_tree_settings: &[(K, V)],
) -> Result<()> {
let e = cstring(engine, "engine")?;
let o = cstring(order_by, "order by")?;
let mt = settings_json(merge_tree_settings)?;
let _guard = self.lib.lock();
unsafe { self.lib.api().engine(self.handle, &e, &o, &mt) }
}
pub fn set_ttl(&mut self, ttl_sql: &str) -> Result<()> {
let t = cstring(ttl_sql, "ttl")?;
let _guard = self.lib.lock();
unsafe { self.lib.api().ttl(self.handle, &t) }
}
pub fn row(&self, format: Format, raw: &[u8]) -> Result<RowResult> {
self.row_with_settings(format, raw, NO_SETTINGS)
}
pub fn row_with_settings<K: AsRef<str>, V: AsRef<str>>(
&self,
format: Format,
raw: &[u8],
settings: &[(K, V)],
) -> Result<RowResult> {
let json = settings_json(settings)?;
let doc: RowDoc = {
let _guard = self.lib.lock();
let out = unsafe { self.lib.api().row(self.handle, format.code(), raw, &json)? };
parse_row_doc(&out)?
};
Ok(row_result_of(doc))
}
pub fn rows<K: AsRef<str>, V: AsRef<str>>(
&self,
format: Format,
body: &[u8],
settings: &[(K, V)],
) -> Result<BatchResult> {
self.rows_through(format, body, settings, EXPORT_NONE, DocFlags::ALL)
}
pub fn rows_export<K: AsRef<str>, V: AsRef<str>>(
&self,
format: Format,
body: &[u8],
settings: &[(K, V)],
export: Option<Format>,
doc_flags: DocFlags,
) -> Result<BatchResult> {
let export_code = export.map_or(EXPORT_NONE, Format::code);
self.rows_through(format, body, settings, export_code, doc_flags)
}
fn rows_through<K: AsRef<str>, V: AsRef<str>>(
&self,
format: Format,
body: &[u8],
settings: &[(K, V)],
export_code: i32,
doc_flags: DocFlags,
) -> Result<BatchResult> {
let json = settings_json(settings)?;
let (doc, payload): (BatchDoc, Option<Vec<u8>>) = {
let _guard = self.lib.lock();
let (out, payload) = unsafe {
self.lib.api().rows(
self.handle,
format.code(),
body,
&json,
export_code,
doc_flags.bits(),
)?
};
(parse_batch_doc(&out)?, payload)
};
let mut res = batch_result_of(doc);
res.payload = payload;
Ok(res)
}
pub fn compile_filter<K: AsRef<str>, V: AsRef<str>>(
&self,
expr_sql: &str,
params: &[(K, V)],
) -> Result<Filter<'_>> {
let e = cstring(expr_sql, "filter expression")?;
let p = settings_json(params)?;
let _guard = self.lib.lock();
let handle = unsafe { self.lib.api().filter_compile(self.handle, &e, &p)? };
Ok(Filter {
schema: self,
handle,
_not_sync: PhantomData,
})
}
pub fn parse_block<K: AsRef<str>, V: AsRef<str>>(
&self,
format: Format,
body: &[u8],
settings: &[(K, V)],
) -> Result<Block<'_>> {
let json = settings_json(settings)?;
let _guard = self.lib.lock();
let handle = unsafe {
self.lib
.api()
.block_parse(self.handle, format.code(), body, &json)?
};
Ok(Block {
schema: self,
handle,
_not_sync: PhantomData,
})
}
}
pub struct Filter<'s> {
schema: &'s Schema,
handle: *mut ChsFilter,
_not_sync: PhantomData<Cell<()>>,
}
impl Filter<'_> {
pub fn schema(&self) -> &Schema {
self.schema
}
pub fn rows<K: AsRef<str>, V: AsRef<str>>(
&self,
format: Format,
body: &[u8],
settings: &[(K, V)],
) -> Result<FilterResult> {
let json = settings_json(settings)?;
let doc: FilterDoc = {
let _guard = self.schema.lib.lock();
let out = unsafe {
self.schema
.lib
.api()
.filter_rows(self.handle, format.code(), body, &json)?
};
crate::doc::filter_doc("e_bare_denormals(&out))?
};
Ok(filter_result_of(doc))
}
pub fn eval(&self, block: &Block<'_>) -> Result<FilterResult> {
if !Arc::ptr_eq(&self.schema.lib, &block.schema.lib) {
return Err(Error::CrossLibrary {
filter_version: self.schema.lib.version().to_string(),
block_version: block.schema.lib.version().to_string(),
});
}
let doc: FilterDoc = {
let _guard = self.schema.lib.lock();
let out = unsafe {
self.schema
.lib
.api()
.filter_eval(self.handle, block.handle)?
};
crate::doc::filter_doc("e_bare_denormals(&out))?
};
Ok(filter_result_of(doc))
}
}
impl Drop for Filter<'_> {
fn drop(&mut self) {
let _guard = self.schema.lib.lock();
unsafe { self.schema.lib.api().filter_free(self.handle) }
}
}
pub struct Block<'s> {
schema: &'s Schema,
handle: *mut ChsBlock,
_not_sync: PhantomData<Cell<()>>,
}
impl Block<'_> {
pub fn schema(&self) -> &Schema {
self.schema
}
}
impl Drop for Block<'_> {
fn drop(&mut self) {
let _guard = self.schema.lib.lock();
unsafe { self.schema.lib.api().block_free(self.handle) }
}
}
impl std::fmt::Debug for Block<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Block")
.field("version", &self.schema.lib.version())
.finish()
}
}
impl std::fmt::Debug for Filter<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Filter")
.field("version", &self.schema.lib.version())
.finish()
}
}
impl Drop for Schema {
fn drop(&mut self) {
let _guard = self.lib.lock();
unsafe { self.lib.api().schema_free(self.handle) }
}
}
impl std::fmt::Debug for Schema {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Schema")
.field("version", &self.lib.version())
.field("columns", &self.columns.len())
.finish()
}
}
pub(crate) fn settings_json<K: AsRef<str>, V: AsRef<str>>(settings: &[(K, V)]) -> Result<CString> {
if settings.is_empty() {
return cstring("{}", "settings");
}
let mut map = serde_json::Map::with_capacity(settings.len());
for (k, v) in settings {
map.insert(
k.as_ref().to_string(),
serde_json::Value::String(v.as_ref().to_string()),
);
}
cstring(&serde_json::Value::Object(map).to_string(), "settings")
}
fn parse_row_doc(bytes: &[u8]) -> Result<RowDoc> {
crate::doc::row_doc("e_bare_denormals(bytes))
}
fn parse_batch_doc(bytes: &[u8]) -> Result<BatchDoc> {
crate::doc::batch_doc("e_bare_denormals(bytes))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn settings_values_are_always_json_strings() {
let json = settings_json(&[(SETTING_NOW_EPOCH_NANOS, "1700000000000000000")]).unwrap();
assert_eq!(
json.to_str().unwrap(),
r#"{"chtypes_now_epoch_nanos":"1700000000000000000"}"#
);
assert_eq!(settings_json(NO_SETTINGS).unwrap().to_str().unwrap(), "{}");
let owned = vec![("input_format_null_as_default".to_string(), "0".to_string())];
assert_eq!(
settings_json(&owned).unwrap().to_str().unwrap(),
r#"{"input_format_null_as_default":"0"}"#
);
}
#[test]
fn a_nanosecond_epoch_survives_verbatim() {
let json = settings_json(&[(SETTING_NOW_EPOCH_NANOS, "1700000000123456789")]).unwrap();
assert!(json.to_str().unwrap().contains("1700000000123456789"));
}
#[test]
fn documents_with_bare_denormals_still_parse() {
let doc = parse_row_doc(
br#"{"outcome":"accepted","cols":[{"name":"f","base":"Float64","src":"input","stored":inf}]}"#,
)
.unwrap();
assert_eq!(doc.cols[0].stored_raw(), "\"inf\"");
}
#[test]
fn a_document_holding_non_utf8_bytes_is_read_exactly_not_repaired() {
let mut doc: Vec<u8> =
br#"{"outcome":"accepted","cols":[{"name":"x","base":"String","src":"input","stored":""#
.to_vec();
doc.extend_from_slice(&[0xc3, b'(']);
doc.extend_from_slice(br#""}]}"#);
let parsed = parse_row_doc(&doc).unwrap();
assert_eq!(
parsed.cols[0].stored_raw().as_bytes(),
[b'"', 0xc3, b'(', b'"']
);
}
}