noth/surrealpatch
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add initial iterator blueprint implementation

Browse source
This commit is contained in:
Tobie Morgan Hitchcock 2022-02-06 01:14:56 +00:00
parent 1600360230
commit 49a62aaba5
37 changed files with 1069 additions and 331 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

111
src/dbs/iterate.rs Normal file
View file

@ -0,0 +1,111 @@
use crate::dbs::Executor;
use crate::dbs::Iterator;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::err::Error;
use crate::sql::array::Array;
use crate::sql::model::Model;
use crate::sql::table::Table;
use crate::sql::thing::Thing;
use crate::sql::value::Value;
use async_recursion::async_recursion;
impl Value {
#[async_recursion]
pub async fn iterate(
self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
ite: &mut Iterator<'_>,
) -> Result<(), Error> {
match self {
Value::Array(v) => v.iterate(ctx, opt, exe, ite).await?,
Value::Model(v) => v.iterate(ctx, opt, exe, ite).await?,
Value::Thing(v) => v.iterate(ctx, opt, exe, ite).await?,
Value::Table(v) => v.iterate(ctx, opt, exe, ite).await?,
v => ite.process(ctx, opt, exe, None, v).await,
}
Ok(())
}
}
impl Array {
#[async_recursion]
pub async fn iterate(
self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
ite: &mut Iterator<'_>,
) -> Result<(), Error> {
for v in self.value.into_iter() {
match v {
Value::Array(v) => v.iterate(ctx, opt, exe, ite).await?,
Value::Model(v) => v.iterate(ctx, opt, exe, ite).await?,
Value::Thing(v) => v.iterate(ctx, opt, exe, ite).await?,
Value::Table(v) => v.iterate(ctx, opt, exe, ite).await?,
v => ite.process(ctx, opt, exe, None, v).await,
}
}
Ok(())
}
}
impl Model {
pub async fn iterate(
self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
ite: &mut Iterator<'_>,
) -> Result<(), Error> {
if ctx.is_ok() {
if let Some(c) = self.count {
for _ in 0..c {
Thing {
tb: self.table.to_string(),
id: xid::new().to_string(),
}
.iterate(ctx, opt, exe, ite)
.await?;
}
}
if let Some(r) = self.range {
for x in r.0..r.1 {
Thing {
tb: self.table.to_string(),
id: x.to_string(),
}
.iterate(ctx, opt, exe, ite)
.await?;
}
}
}
Ok(())
}
}
impl Thing {
pub async fn iterate(
self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
ite: &mut Iterator<'_>,
) -> Result<(), Error> {
todo!()
}
}
impl Table {
pub async fn iterate(
self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
ite: &mut Iterator<'_>,
) -> Result<(), Error> {
todo!()
}
}

372
src/dbs/iterator.rs
View file

@ -1,14 +1,13 @@
use crate::ctx::Canceller;
use crate::ctx::Context;
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::array::Array;
use crate::sql::cond::Cond;
use crate::sql::data::Data;
use crate::sql::field::Fields;
use crate::sql::group::Groups;
use crate::sql::limit::Limit;
use crate::sql::model::Model;
use crate::sql::object::Object;
use crate::sql::order::Orders;
use crate::sql::split::Splits;
use crate::sql::start::Start;
@ -17,258 +16,209 @@ use crate::sql::thing::Thing;
use crate::sql::value::Value;
use crate::sql::version::Version;
use std::mem;
use tokio::sync::mpsc;
use tokio::sync::mpsc::UnboundedSender;
use xid;
pub type Channel = UnboundedSender<Value>;
#[derive(Default)]
pub struct Iterator<'a> {
ok: bool,
// Iterator status
run: Canceller,
// Iterator runtime error
error: Option<Error>,
// Iterator input values
readies: Vec<Value>,
// Iterator output results
results: Vec<Value>,
// Iterate options
pub parallel: bool,
// Underlying statement
pub stmt: Statement<'a>,
// Iterator options
pub into: Option<&'a Table>,
pub expr: Option<&'a Fields>,
pub data: Option<&'a Data>,
pub cond: Option<&'a Cond>,
pub split: Option<&'a Splits>,
pub group: Option<&'a Groups>,
pub order: Option<&'a Orders>,
pub limit: Option<&'a Limit>,
pub start: Option<&'a Start>,
pub version: Option<&'a Version>,
// Iterator runtime error
error: Option<Error>,
// Iterator output results
results: Vec<Value>,
}
impl<'a> Iterator<'a> {
pub fn new() -> Iterator<'a> {
Iterator {
ok: true,
..Iterator::default()
}
Iterator::default()
}
fn check(&self, ctx: &Runtime) -> bool {
self.ok && ctx.is_ok()
// Prepares a value for processing
pub fn prepare(&mut self, val: Value) {
self.readies.push(val)
}
pub fn process_table(&mut self, ctx: &Runtime, exe: &mut Executor, val: Table) {
// Check basic permissions
self.process_perms(ctx, exe);
// Loop over all table keys
// - Process record
// Create a new record for processing
pub fn produce(&mut self, val: Table) {
self.prepare(Value::Thing(Thing {
tb: val.name.to_string(),
id: xid::new().to_string(),
}))
}
pub fn process_thing(&mut self, ctx: &Runtime, exe: &mut Executor, val: Thing) {
// Check basic permissions
self.process_perms(ctx, exe);
// Check current context
if self.check(ctx) {
// Process record
// self.process(ctx, exe);
pub async fn output(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
) -> Result<Value, Error> {
// Log the statement
trace!("Iterating {}", self.stmt);
// Enable context override
let mut ctx = Context::new(&ctx);
self.run = ctx.add_cancel();
let ctx = ctx.freeze();
// Process prepared values
self.iterate(&ctx, opt, exe).await?;
// Return any document errors
if let Some(e) = self.error.take() {
return Err(e);
}
// Process any SPLIT clause
self.output_split(&ctx, opt, exe);
// Process any GROUP clause
self.output_group(&ctx, opt, exe);
// Process any ORDER clause
self.output_order(&ctx, opt, exe);
// Process any START clause
self.output_start(&ctx, opt, exe);
// Process any LIMIT clause
self.output_limit(&ctx, opt, exe);
// Output the results
Ok(mem::take(&mut self.results).into())
}
pub fn process_model(&mut self, ctx: &Runtime, exe: &mut Executor, val: Model) {
// Check basic permissions
self.process_perms(ctx, exe);
// Process count based model
if val.count.is_some() {
let c = val.count.unwrap();
for _ in 0..c {
// Check current context
if self.check(ctx) {
// Process record
self.process(
ctx,
exe,
Value::from(Thing {
tb: val.table.to_string(),
id: xid::new().to_string(),
}),
);
}
}
}
// Process range based model
if val.range.is_some() {
let r = val.range.unwrap();
for x in r.0..r.1 {
// Check current context
if self.check(ctx) {
// Process record
self.process(
ctx,
exe,
Value::from(Thing {
tb: val.table.to_string(),
id: x.to_string(),
}),
);
}
}
}
}
pub fn process_array(&mut self, ctx: &Runtime, exe: &mut Executor, val: Array) {
// Check basic permissions
self.process_perms(ctx, exe);
// Loop over query result array
for v in val.value.into_iter() {
// Check current context
if self.check(ctx) {
// Process item
match v {
Value::Thing(v) => self.process_thing(ctx, exe, v),
Value::Object(v) => self.process_object(ctx, exe, v),
v => self.process(ctx, exe, v),
}
}
}
}
pub fn process_object(&mut self, ctx: &Runtime, exe: &mut Executor, val: Object) {
// Check basic permissions
self.process_perms(ctx, exe);
// Check current context
if self.check(ctx) {
// Loop over query result array
self.process(ctx, exe, val.into())
}
}
pub fn process_value(&mut self, ctx: &Runtime, exe: &mut Executor, val: Value) {
// Check basic permissions
self.process_perms(ctx, exe);
// Loop over query result array
// self.process(ctx, exe, val)
// - IF value is THING then process record
// - IF value.id is THING then process record
// - ELSE process as object
match val {
Value::Thing(v) => self.process_thing(ctx, exe, v),
Value::Object(v) => self.process_object(ctx, exe, v),
v => self.process(ctx, exe, v),
}
}
fn process(&mut self, ctx: &Runtime, exe: &mut Executor, val: Value) {
// 1. Setup a new document
// 2. Check for any errors
// 3. Append the result
let res = Some(val);
// If an error was received from the
// worker, then set the error if no
// previous iterator error has occured.
if self.check(ctx) == false {
return;
}
// Otherwise add the received result
// to the iterator result slice so
// that it is ready for processing.
if let Some(r) = res {
self.results.push(r);
}
// The statement does not have a limit
// expression specified, so therefore
// we need to load all data before
// stopping the iterator.
if self.limit.is_none() {
return;
}
// If the statement specified a GROUP
// BY expression, then we need to load
// all data from all sources before
// stopping the iterator.
if self.group.is_some() {
return;
}
// If the statement specified an ORDER
// BY expression, then we need to load
// all data from all sources before
// stopping the iterator.
if self.order.is_some() {
return;
}
// Otherwise we can stop the iterator
// early, if we have the necessary
// number of records specified in the
// query statement.
if let Some(l) = self.limit {
if let Some(s) = self.start {
if self.results.len() == l.0 + s.0 {
self.ok = false
}
} else {
if self.results.len() == l.0 {
self.ok = false
}
}
}
}
fn process_perms(&self, ctx: &Runtime, exe: &Executor) {}
fn process_split(&mut self, ctx: &Runtime, exe: &Executor) {
#[inline]
fn output_split(&mut self, ctx: &Runtime, opt: &Options<'_>, exe: &Executor) {
if self.split.is_some() {
// Ignore
}
}
fn process_group(&mut self, ctx: &Runtime, exe: &Executor) {
#[inline]
fn output_group(&mut self, ctx: &Runtime, opt: &Options<'_>, exe: &Executor) {
if self.group.is_some() {
// Ignore
}
}
fn process_order(&mut self, ctx: &Runtime, exe: &Executor) {
#[inline]
fn output_order(&mut self, ctx: &Runtime, opt: &Options<'_>, exe: &Executor) {
if self.order.is_some() {
// Ignore
}
}
fn process_start(&mut self, ctx: &Runtime, exe: &Executor) {
#[inline]
fn output_start(&mut self, ctx: &Runtime, opt: &Options<'_>, exe: &Executor) {
if let Some(v) = self.start {
let s = v.0 as usize;
self.results = mem::take(&mut self.results).into_iter().skip(s).collect();
self.results = mem::take(&mut self.results).into_iter().skip(v.0).collect();
}
}
fn process_limit(&mut self, ctx: &Runtime, exe: &Executor) {
#[inline]
fn output_limit(&mut self, ctx: &Runtime, opt: &Options<'_>, exe: &Executor) {
if let Some(v) = self.limit {
let l = v.0 as usize;
self.results = mem::take(&mut self.results).into_iter().take(l).collect();
self.results = mem::take(&mut self.results).into_iter().take(v.0).collect();
}
}
pub fn output(&mut self, ctx: &Runtime, exe: &Executor) -> Result<Value, Error> {
// Return any errors
if let Some(e) = self.error.take() {
return Err(e);
async fn iterate(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
) -> Result<(), Error> {
match self.parallel {
// Run statements in parallel
true => {
// Create an unbounded channel
let (_, mut rx) = mpsc::unbounded_channel();
// Process all prepared values
for _ in mem::take(&mut self.readies) {
todo!();
}
// Process all processed values
while let Some(v) = rx.recv().await {
self.process(&ctx, opt, exe, None, v).await;
}
// Everything processed ok
Ok(())
}
// Run statements sequentially
false => {
// Process all prepared values
for v in mem::take(&mut self.readies) {
v.iterate(ctx, opt, exe, self).await?;
}
// Everything processed ok
Ok(())
}
}
}
pub async fn process(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
thg: Option<Thing>,
val: Value,
) {
// Check current context
if ctx.is_done() {
return;
}
// Setup a new document
let mut doc = Document::new(thg, val);
// Process the document
let res = match self.stmt {
Statement::Select(_) => doc.select(ctx, opt, exe, &self.stmt).await,
Statement::Create(_) => doc.create(ctx, opt, exe, &self.stmt).await,
Statement::Update(_) => doc.update(ctx, opt, exe, &self.stmt).await,
Statement::Relate(_) => doc.relate(ctx, opt, exe, &self.stmt).await,
Statement::Delete(_) => doc.delete(ctx, opt, exe, &self.stmt).await,
Statement::Insert(_) => doc.insert(ctx, opt, exe, &self.stmt).await,
_ => unreachable!(),
};
// Process the result
match res {
Err(Error::IgnoreError) => {
self.run.cancel();
return;
}
Err(e) => {
self.error = Some(e);
self.run.cancel();
return;
}
Ok(v) => self.results.push(v),
}
// Check if we can exit
if self.group.is_none() {
if self.order.is_none() {
if let Some(l) = self.limit {
if let Some(s) = self.start {
if self.results.len() == l.0 + s.0 {
self.run.cancel()
}
} else {
if self.results.len() == l.0 {
self.run.cancel()
}
}
}
}
}
// Process SPLIT clause
self.process_split(ctx, exe);
// Process GROUP clause
self.process_group(ctx, exe);
// Process ORDER clause
self.process_order(ctx, exe);
// Process START clause
self.process_start(ctx, exe);
// Process LIMIT clause
self.process_limit(ctx, exe);
// Output the results
Ok(mem::take(&mut self.results).into())
}
}

3
src/dbs/mod.rs
View file

@ -2,11 +2,13 @@ mod auth;
mod dbs;
mod executor;
mod export;
mod iterate;
mod iterator;
mod options;
mod response;
mod runtime;
mod session;
mod statement;
mod variables;
pub use self::auth::*;
@ -17,6 +19,7 @@ pub use self::options::*;
pub use self::response::*;
pub use self::runtime::*;
pub use self::session::*;
pub use self::statement::*;
pub use self::variables::*;
#[cfg(test)]

74
src/dbs/statement.rs Normal file
View file

@ -0,0 +1,74 @@
use crate::sql::statements::create::CreateStatement;
use crate::sql::statements::delete::DeleteStatement;
use crate::sql::statements::insert::InsertStatement;
use crate::sql::statements::relate::RelateStatement;
use crate::sql::statements::select::SelectStatement;
use crate::sql::statements::update::UpdateStatement;
use std::fmt;
#[derive(Debug)]
pub enum Statement<'a> {
None,
Select(&'a SelectStatement),
Create(&'a CreateStatement),
Update(&'a UpdateStatement),
Relate(&'a RelateStatement),
Delete(&'a DeleteStatement),
Insert(&'a InsertStatement),
}
impl Default for Statement<'_> {
fn default() -> Self {
Statement::None
}
}
impl<'a> From<&'a SelectStatement> for Statement<'a> {
fn from(v: &'a SelectStatement) -> Self {
Statement::Select(v)
}
}
impl<'a> From<&'a CreateStatement> for Statement<'a> {
fn from(v: &'a CreateStatement) -> Self {
Statement::Create(v)
}
}
impl<'a> From<&'a UpdateStatement> for Statement<'a> {
fn from(v: &'a UpdateStatement) -> Self {
Statement::Update(v)
}
}
impl<'a> From<&'a RelateStatement> for Statement<'a> {
fn from(v: &'a RelateStatement) -> Self {
Statement::Relate(v)
}
}
impl<'a> From<&'a DeleteStatement> for Statement<'a> {
fn from(v: &'a DeleteStatement) -> Self {
Statement::Delete(v)
}
}
impl<'a> From<&'a InsertStatement> for Statement<'a> {
fn from(v: &'a InsertStatement) -> Self {
Statement::Insert(v)
}
}
impl<'a> fmt::Display for Statement<'a> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Statement::Select(v) => write!(f, "{}", v),
Statement::Create(v) => write!(f, "{}", v),
Statement::Update(v) => write!(f, "{}", v),
Statement::Relate(v) => write!(f, "{}", v),
Statement::Delete(v) => write!(f, "{}", v),
Statement::Insert(v) => write!(f, "{}", v),
_ => unreachable!(),
}
}
}

38
src/doc/admit.rs Normal file
View file

@ -0,0 +1,38 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn admit(
&self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<(), Error> {
match self.id {
Some(_) => Ok(()),
None => match stm {
Statement::Create(_) => Err(Error::CreateStatementError {
value: self.initial.clone(),
}),
Statement::Update(_) => Err(Error::UpdateStatementError {
value: self.initial.clone(),
}),
Statement::Relate(_) => Err(Error::RelateStatementError {
value: self.initial.clone(),
}),
Statement::Delete(_) => Err(Error::DeleteStatementError {
value: self.initial.clone(),
}),
Statement::Insert(_) => Err(Error::InsertStatementError {
value: self.initial.clone(),
}),
_ => unreachable!(),
},
}
}
}

17
src/doc/allow.rs
View file

@ -1 +1,18 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn allow(
&self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<(), Error> {
Ok(())
}
}

32
src/doc/check.rs
View file

@ -1 +1,33 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn check(
&self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<(), Error> {
// Extract statement clause
let cond = match stm {
Statement::Select(stm) => stm.cond.as_ref(),
Statement::Update(stm) => stm.cond.as_ref(),
_ => unreachable!(),
};
// Match clause
match cond {
Some(v) => {
match v.expr.compute(ctx, opt, exe, Some(&self.current)).await?.is_truthy() {
false => Err(Error::IgnoreError),
true => Ok(()),
}
}
None => Ok(()),
}
}
}

27
src/doc/compute.rs Normal file
View file

@ -0,0 +1,27 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::value::Value;
impl Document {
pub async fn compute(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<Value, Error> {
match stm {
Statement::Select(_) => self.select(ctx, opt, exe, stm).await,
Statement::Create(_) => self.create(ctx, opt, exe, stm).await,
Statement::Update(_) => self.update(ctx, opt, exe, stm).await,
Statement::Relate(_) => self.relate(ctx, opt, exe, stm).await,
Statement::Delete(_) => self.delete(ctx, opt, exe, stm).await,
Statement::Insert(_) => self.insert(ctx, opt, exe, stm).await,
_ => unreachable!(),
}
}
}

35
src/doc/create.rs
View file

@ -1 +1,36 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::value::Value;
impl Document {
pub async fn create(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<Value, Error> {
// Check value type
self.admit(ctx, opt, exe, stm).await?;
// Merge record data
self.merge(ctx, opt, exe, stm).await?;
// Check if allowed
self.allow(ctx, opt, exe, stm).await?;
// Store index data
self.index(ctx, opt, exe, stm).await?;
// Store record data
self.store(ctx, opt, exe, stm).await?;
// Run table queries
self.table(ctx, opt, exe, stm).await?;
// Run lives queries
self.lives(ctx, opt, exe, stm).await?;
// Run event queries
self.event(ctx, opt, exe, stm).await?;
// Yield document
self.pluck(ctx, opt, exe, stm).await
}
}

37
src/doc/delete.rs
View file

@ -1 +1,38 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::value::Value;
impl Document {
pub async fn delete(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<Value, Error> {
// Check value type
self.admit(ctx, opt, exe, stm).await?;
// Check where clause
self.check(ctx, opt, exe, stm).await?;
// Check if allowed
self.allow(ctx, opt, exe, stm).await?;
// Erase document
self.erase(ctx, opt, exe, stm).await?;
// Store index data
self.index(ctx, opt, exe, stm).await?;
// Store record data
self.store(ctx, opt, exe, stm).await?;
// Run table queries
self.table(ctx, opt, exe, stm).await?;
// Run lives queries
self.lives(ctx, opt, exe, stm).await?;
// Run event queries
self.event(ctx, opt, exe, stm).await?;
// Yield document
self.pluck(ctx, opt, exe, stm).await
}
}

2
src/doc/doc.rs
View file

@ -1,2 +0,0 @@
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct Document {}

19
src/doc/document.rs Normal file
View file

@ -0,0 +1,19 @@
use crate::sql::thing::Thing;
use crate::sql::value::Value;
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct Document {
pub(super) id: Option<Thing>,
pub(super) current: Value,
pub(super) initial: Value,
}
impl Document {
pub fn new(id: Option<Thing>, val: Value) -> Document {
Document {
id,
current: val.clone(),
initial: val,
}
}
}

18
src/doc/erase.rs Normal file
View file

@ -0,0 +1,18 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn erase(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<(), Error> {
self.current.clear(ctx, opt, exe).await
}
}

17
src/doc/event.rs
View file

@ -1 +1,18 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn event(
&self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<(), Error> {
Ok(())
}
}

17
src/doc/index.rs
View file

@ -1 +1,18 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn index(
&self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<(), Error> {
Ok(())
}
}

18
src/doc/insert.rs
View file

@ -1 +1,19 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::value::Value;
impl Document {
pub async fn insert(
&mut self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<Value, Error> {
todo!()
}
}

17
src/doc/lives.rs
View file

@ -1 +1,18 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn lives(
&self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<(), Error> {
Ok(())
}
}

64
src/doc/merge.rs
View file

@ -1 +1,65 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::data::Data;
use crate::sql::operator::Operator;
use crate::sql::value::Value;
impl Document {
pub async fn merge(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<(), Error> {
// Get the ID reference
let id = self.id.as_ref();
// Extract statement clause
let data = match stm {
Statement::Create(stm) => stm.data.as_ref(),
Statement::Update(stm) => stm.data.as_ref(),
_ => unreachable!(),
};
// Set default field values
self.current.def(ctx, opt, exe, id).await?;
// Check for a data clause
match data {
// The statement has a data clause
Some(v) => match v {
Data::SetExpression(x) => {
for x in x.iter() {
let v = x.2.compute(ctx, opt, exe, Some(&self.current)).await?;
match x.1 {
Operator::Equal => match v {
Value::Void => self.current.del(ctx, opt, exe, &x.0).await?,
_ => self.current.set(ctx, opt, exe, &x.0, v).await?,
},
Operator::Inc => self.current.increment(ctx, opt, exe, &x.0, v).await?,
Operator::Dec => self.current.decrement(ctx, opt, exe, &x.0, v).await?,
_ => unreachable!(),
}
}
}
Data::PatchExpression(v) => self.current.patch(ctx, opt, exe, v).await?,
Data::MergeExpression(v) => self.current.merge(ctx, opt, exe, v).await?,
Data::ReplaceExpression(v) => self.current.replace(ctx, opt, exe, v).await?,
Data::ContentExpression(v) => self.current.replace(ctx, opt, exe, v).await?,
_ => unreachable!(),
},
// No data clause has been set
None => (),
};
// Set default field values
self.current.def(ctx, opt, exe, id).await?;
// Set ASSERT and VALUE clauses
// todo!();
// Delete non-defined FIELDs
// todo!();
// Carry on
Ok(())
}
}

10
src/doc/mod.rs
View file

@ -1,10 +1,13 @@
pub use self::doc::*;
pub use self::document::*;
mod admit;
mod allow;
mod check;
mod compute;
mod create;
mod delete;
mod doc;
mod document;
mod erase;
mod event;
mod grant;
mod index;
@ -12,8 +15,9 @@ mod insert;
mod lives;
mod merge;
mod perms;
mod pluck;
mod relate;
mod select;
mod store;
mod table;
mod update;
mod r#yield;

86
src/doc/pluck.rs Normal file
View file

@ -0,0 +1,86 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::field::Field;
use crate::sql::idiom::Idiom;
use crate::sql::output::Output;
use crate::sql::value::Value;
impl Document {
pub async fn pluck(
&self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<Value, Error> {
// Extract statement clause
let expr = match stm {
Statement::Select(_) => None,
Statement::Create(stm) => stm.output.as_ref().or(Some(&Output::After)),
Statement::Update(stm) => stm.output.as_ref().or(Some(&Output::After)),
Statement::Relate(stm) => stm.output.as_ref().or(Some(&Output::After)),
Statement::Delete(stm) => stm.output.as_ref().or(Some(&Output::None)),
Statement::Insert(stm) => stm.output.as_ref().or(Some(&Output::After)),
_ => unreachable!(),
};
// Ensure futures are run
let opt = &opt.futures(true);
// Match clause
match expr {
Some(v) => match v {
Output::None => Err(Error::IgnoreError),
Output::Null => Ok(Value::Null),
Output::Diff => Ok(self.initial.diff(&self.current, Idiom::default()).into()),
Output::After => self.current.compute(ctx, opt, exe, Some(&self.current)).await,
Output::Before => self.initial.compute(ctx, opt, exe, Some(&self.initial)).await,
Output::Fields(v) => {
let mut out = match v.all() {
true => self.current.compute(ctx, opt, exe, Some(&self.current)).await?,
false => Value::base(),
};
for v in v.iter() {
match v {
Field::All => (),
Field::Alone(v) => {
let x = v.compute(ctx, opt, exe, Some(&self.current)).await?;
out.set(ctx, opt, exe, &v.to_idiom(), x).await?;
}
Field::Alias(v, i) => {
let x = v.compute(ctx, opt, exe, Some(&self.current)).await?;
out.set(ctx, opt, exe, &i, x).await?;
}
}
}
Ok(out)
}
},
None => match stm {
Statement::Select(stm) => {
let mut out = match stm.expr.all() {
true => self.current.compute(ctx, opt, exe, Some(&self.current)).await?,
false => Value::base(),
};
for v in stm.expr.iter() {
match v {
Field::All => (),
Field::Alone(v) => {
let x = v.compute(ctx, opt, exe, Some(&self.current)).await?;
out.set(ctx, opt, exe, &v.to_idiom(), x).await?;
}
Field::Alias(v, i) => {
let x = v.compute(ctx, opt, exe, Some(&self.current)).await?;
out.set(ctx, opt, exe, &i, x).await?;
}
}
}
Ok(out)
}
_ => unreachable!(),
},
}
}
}

37
src/doc/relate.rs
View file

@ -1 +1,38 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::value::Value;
impl Document {
pub async fn relate(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<Value, Error> {
// Check value type
self.admit(ctx, opt, exe, stm).await?;
// Check if allowed
self.allow(ctx, opt, exe, stm).await?;
// Merge record data
self.merge(ctx, opt, exe, stm).await?;
// Check if allowed
self.allow(ctx, opt, exe, stm).await?;
// Store index data
self.index(ctx, opt, exe, stm).await?;
// Store record data
self.store(ctx, opt, exe, stm).await?;
// Run table queries
self.table(ctx, opt, exe, stm).await?;
// Run lives queries
self.lives(ctx, opt, exe, stm).await?;
// Run event queries
self.event(ctx, opt, exe, stm).await?;
// Yield document
self.pluck(ctx, opt, exe, stm).await
}
}

23
src/doc/select.rs
View file

@ -1 +1,24 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::value::Value;
impl Document {
pub async fn select(
&self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<Value, Error> {
// Check where clause
self.check(ctx, opt, exe, stm).await?;
// Check if allowed
self.allow(ctx, opt, exe, stm).await?;
// Yield document
self.pluck(ctx, opt, exe, stm).await
}
}

18
src/doc/store.rs Normal file
View file

@ -0,0 +1,18 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn store(
&self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<(), Error> {
Ok(())
}
}

17
src/doc/table.rs
View file

@ -1 +1,18 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub async fn table(
&self,
_ctx: &Runtime,
_opt: &Options<'_>,
_exe: &Executor<'_>,
_stm: &Statement<'_>,
) -> Result<(), Error> {
Ok(())
}
}

39
src/doc/update.rs
View file

@ -1 +1,40 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
use crate::sql::value::Value;
impl Document {
pub async fn update(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
stm: &Statement<'_>,
) -> Result<Value, Error> {
// Check value type
self.admit(ctx, opt, exe, stm).await?;
// Check where clause
self.check(ctx, opt, exe, stm).await?;
// Check if allowed
self.allow(ctx, opt, exe, stm).await?;
// Merge record data
self.merge(ctx, opt, exe, stm).await?;
// Check if allowed
self.allow(ctx, opt, exe, stm).await?;
// Store index data
self.index(ctx, opt, exe, stm).await?;
// Store record data
self.store(ctx, opt, exe, stm).await?;
// Run table queries
self.table(ctx, opt, exe, stm).await?;
// Run lives queries
self.lives(ctx, opt, exe, stm).await?;
// Run event queries
self.event(ctx, opt, exe, stm).await?;
// Yield document
self.pluck(ctx, opt, exe, stm).await
}
}

1
src/doc/yield.rs
View file

@ -1 +0,0 @@

3
src/err/mod.rs
View file

@ -142,6 +142,9 @@ pub enum Error {
thing: Thing,
},
#[error("Conditional clause is not truthy")]
IgnoreError,
#[error("Key encoding error: {0}")]
EncodeError(#[from] EncodeError),

15
src/sql/field.rs
View file

@ -13,6 +13,21 @@ use std::fmt;
pub struct Fields(pub Vec<Field>);
impl Fields {
pub fn all(&self) -> bool {
self.0.iter().any(|v| match v {
Field::All => true,
_ => false,
})
}
pub fn iter(&self) -> impl Iterator<Item = &Field> {
self.0.iter()
}
pub fn other(&self) -> impl Iterator<Item = &Field> {
self.0.iter().filter(|v| match v {
Field::All => false,
_ => true,
})
}
pub fn single(&self) -> Option<Idiom> {
match self.0.len() {
1 => match self.0.first() {

29
src/sql/statements/create.rs
View file

@ -3,6 +3,7 @@ use crate::dbs::Iterator;
use crate::dbs::Level;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::err::Error;
use crate::sql::comment::shouldbespace;
use crate::sql::data::{data, Data};
@ -39,28 +40,18 @@ impl CreateStatement {
exe.check(opt, Level::No)?;
// Create a new iterator
let mut i = Iterator::new();
// Pass in statement config
i.data = self.data.as_ref();
// Pass in current statement
i.stmt = Statement::from(self);
// Ensure futures are stored
let opt = &opt.futures(false);
// Loop over the create targets
for w in self.what.0.iter() {
match w.compute(ctx, opt, exe, doc).await? {
Value::Table(v) => {
i.process_table(ctx, exe, v);
}
Value::Thing(v) => {
i.process_thing(ctx, exe, v);
}
Value::Model(v) => {
i.process_model(ctx, exe, v);
}
Value::Array(v) => {
i.process_array(ctx, exe, v);
}
Value::Object(v) => {
i.process_object(ctx, exe, v);
}
let v = w.compute(ctx, opt, exe, doc).await?;
match v {
Value::Table(v) => i.produce(v),
Value::Thing(_) => i.prepare(v),
Value::Model(_) => i.prepare(v),
Value::Array(_) => i.prepare(v),
v => {
return Err(Error::CreateStatementError {
value: v,
@ -69,7 +60,7 @@ impl CreateStatement {
};
}
// Output the results
i.output(ctx, exe)
i.output(ctx, opt, exe).await
}
}

29
src/sql/statements/delete.rs
View file

@ -3,6 +3,7 @@ use crate::dbs::Iterator;
use crate::dbs::Level;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::err::Error;
use crate::sql::comment::shouldbespace;
use crate::sql::cond::{cond, Cond};
@ -40,28 +41,18 @@ impl DeleteStatement {
exe.check(opt, Level::No)?;
// Create a new iterator
let mut i = Iterator::new();
// Pass in statement config
i.cond = self.cond.as_ref();
// Pass in current statement
i.stmt = Statement::from(self);
// Ensure futures are stored
let opt = &opt.futures(false);
// Loop over the delete targets
for w in self.what.0.iter() {
match w.compute(ctx, opt, exe, doc).await? {
Value::Table(v) => {
i.process_table(ctx, exe, v);
}
Value::Thing(v) => {
i.process_thing(ctx, exe, v);
}
Value::Model(v) => {
i.process_model(ctx, exe, v);
}
Value::Array(v) => {
i.process_array(ctx, exe, v);
}
Value::Object(v) => {
i.process_object(ctx, exe, v);
}
let v = w.compute(ctx, opt, exe, doc).await?;
match v {
Value::Table(_) => i.prepare(v),
Value::Thing(_) => i.prepare(v),
Value::Model(_) => i.prepare(v),
Value::Array(_) => i.prepare(v),
v => {
return Err(Error::DeleteStatementError {
value: v,
@ -70,7 +61,7 @@ impl DeleteStatement {
};
}
// Output the results
i.output(ctx, exe)
i.output(ctx, opt, exe).await
}
}

31
src/sql/statements/insert.rs
View file

@ -3,6 +3,7 @@ use crate::dbs::Iterator;
use crate::dbs::Level;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::err::Error;
use crate::sql::comment::shouldbespace;
use crate::sql::data::{single, update, values, Data};
@ -43,9 +44,8 @@ impl InsertStatement {
exe.check(opt, Level::No)?;
// Create a new iterator
let mut i = Iterator::new();
// Pass in statement config
i.into = Some(&self.into);
i.data = Some(&self.data);
// Pass in current statement
i.stmt = Statement::from(self);
// Ensure futures are stored
let opt = &opt.futures(false);
// Parse the expression
@ -53,23 +53,22 @@ impl InsertStatement {
Data::ValuesExpression(_) => {
todo!() // TODO: loop over each
}
Data::SingleExpression(v) => match v.compute(ctx, opt, exe, doc).await? {
Value::Array(v) => {
i.process_array(ctx, exe, v);
Data::SingleExpression(v) => {
let v = v.compute(ctx, opt, exe, doc).await?;
match v {
Value::Array(v) => v.value.into_iter().for_each(|v| i.prepare(v)),
Value::Object(_) => i.prepare(v),
v => {
return Err(Error::InsertStatementError {
value: v,
})
}
}
Value::Object(v) => {
i.process_object(ctx, exe, v);
}
v => {
return Err(Error::InsertStatementError {
value: v,
})
}
},
}
_ => unreachable!(),
}
// Output the results
i.output(ctx, exe)
i.output(ctx, opt, exe).await
}
}

29
src/sql/statements/relate.rs
View file

@ -3,6 +3,7 @@ use crate::dbs::Iterator;
use crate::dbs::Level;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::err::Error;
use crate::sql::comment::mightbespace;
use crate::sql::comment::shouldbespace;
@ -46,26 +47,18 @@ impl RelateStatement {
exe.check(opt, Level::No)?;
// Create a new iterator
let mut i = Iterator::new();
// Pass in current statement
i.stmt = Statement::from(self);
// Ensure futures are stored
let opt = &opt.futures(false);
// Loop over the select targets
for f in self.from.0.iter() {
match f.compute(ctx, opt, exe, doc).await? {
Value::Table(v) => {
i.process_table(ctx, exe, v);
}
Value::Thing(v) => {
i.process_thing(ctx, exe, v);
}
Value::Model(v) => {
i.process_model(ctx, exe, v);
}
Value::Array(v) => {
i.process_array(ctx, exe, v);
}
Value::Object(v) => {
i.process_object(ctx, exe, v);
}
for w in self.from.0.iter() {
let v = w.compute(ctx, opt, exe, doc).await?;
match v {
Value::Table(_) => i.prepare(v),
Value::Thing(_) => i.prepare(v),
Value::Model(_) => i.prepare(v),
Value::Array(_) => i.prepare(v),
v => {
return Err(Error::RelateStatementError {
value: v,
@ -74,7 +67,7 @@ impl RelateStatement {
};
}
// Output the results
i.output(ctx, exe)
i.output(ctx, opt, exe).await
}
}

33
src/sql/statements/select.rs
View file

@ -3,6 +3,7 @@ use crate::dbs::Iterator;
use crate::dbs::Level;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::err::Error;
use crate::sql::comment::shouldbespace;
use crate::sql::cond::{cond, Cond};
@ -74,9 +75,9 @@ impl SelectStatement {
exe.check(opt, Level::No)?;
// Create a new iterator
let mut i = Iterator::new();
// Pass in current statement
i.stmt = Statement::from(self);
// Pass in statement config
i.expr = Some(&self.expr);
i.cond = self.cond.as_ref();
i.split = self.split.as_ref();
i.group = self.group.as_ref();
i.order = self.order.as_ref();
@ -88,29 +89,17 @@ impl SelectStatement {
let opt = &opt.version(self.version.as_ref());
// Loop over the select targets
for w in self.what.0.iter() {
match w.compute(ctx, opt, exe, doc).await? {
Value::Table(v) => {
i.process_table(ctx, exe, v);
}
Value::Thing(v) => {
i.process_thing(ctx, exe, v);
}
Value::Model(v) => {
i.process_model(ctx, exe, v);
}
Value::Array(v) => {
i.process_array(ctx, exe, v);
}
Value::Object(v) => {
i.process_object(ctx, exe, v);
}
v => {
i.process_value(ctx, exe, v);
}
let v = w.compute(ctx, opt, exe, doc).await?;
match v {
Value::Table(_) => i.prepare(v),
Value::Thing(_) => i.prepare(v),
Value::Model(_) => i.prepare(v),
Value::Array(_) => i.prepare(v),
v => i.prepare(v),
};
}
// Output the results
i.output(ctx, exe)
i.output(ctx, opt, exe).await
}
}

30
src/sql/statements/update.rs
View file

@ -3,6 +3,7 @@ use crate::dbs::Iterator;
use crate::dbs::Level;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::dbs::Statement;
use crate::err::Error;
use crate::sql::comment::shouldbespace;
use crate::sql::cond::{cond, Cond};
@ -42,29 +43,18 @@ impl UpdateStatement {
exe.check(opt, Level::No)?;
// Create a new iterator
let mut i = Iterator::new();
// Pass in statement config
i.data = self.data.as_ref();
i.cond = self.cond.as_ref();
// Pass in current statement
i.stmt = Statement::from(self);
// Ensure futures are stored
let opt = &opt.futures(false);
// Loop over the update targets
for w in self.what.0.iter() {
match w.compute(ctx, opt, exe, doc).await? {
Value::Table(v) => {
i.process_table(ctx, exe, v);
}
Value::Thing(v) => {
i.process_thing(ctx, exe, v);
}
Value::Model(v) => {
i.process_model(ctx, exe, v);
}
Value::Array(v) => {
i.process_array(ctx, exe, v);
}
Value::Object(v) => {
i.process_object(ctx, exe, v);
}
let v = w.compute(ctx, opt, exe, doc).await?;
match v {
Value::Table(_) => i.prepare(v),
Value::Thing(_) => i.prepare(v),
Value::Model(_) => i.prepare(v),
Value::Array(_) => i.prepare(v),
v => {
return Err(Error::UpdateStatementError {
value: v,
@ -73,7 +63,7 @@ impl UpdateStatement {
};
}
// Output the results
i.output(ctx, exe)
i.output(ctx, opt, exe).await
}
}

43
src/sql/value/def.rs Normal file
View file

@ -0,0 +1,43 @@
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Runtime;
use crate::err::Error;
use crate::sql::idiom::Idiom;
use crate::sql::part::Part;
use crate::sql::thing::Thing;
use crate::sql::value::Value;
use once_cell::sync::Lazy;
static RID: Lazy<Idiom> = Lazy::new(|| Idiom {
parts: vec![Part::from("id")],
});
static MTB: Lazy<Idiom> = Lazy::new(|| Idiom {
parts: vec![Part::from("meta"), Part::from("tb")],
});
static MID: Lazy<Idiom> = Lazy::new(|| Idiom {
parts: vec![Part::from("meta"), Part::from("id")],
});
impl Value {
pub async fn def(
&mut self,
ctx: &Runtime,
opt: &Options<'_>,
exe: &Executor<'_>,
val: Option<&Thing>,
) -> Result<(), Error> {
match val {
Some(id) => {
let id = id.clone();
let md = id.clone();
self.set(ctx, opt, exe, &RID, id.into()).await?;
self.set(ctx, opt, exe, &MTB, md.tb.into()).await?;
self.set(ctx, opt, exe, &MID, md.id.into()).await?;
Ok(())
}
None => unreachable!(),
}
}
}

1
src/sql/value/mod.rs
View file

@ -3,6 +3,7 @@ pub use self::value::*;
pub mod array;
pub mod clear;
pub mod decrement;
pub mod def;
pub mod del;
pub mod diff;
pub mod fetch;

8
src/sql/value/value.rs
View file

@ -387,6 +387,14 @@ impl From<Option<String>> for Value {
}
impl Value {
// -----------------------------------
// Initial record value
// -----------------------------------
pub fn base() -> Self {
Value::Object(Object::default())
}
// -----------------------------------
// Builtin types
// -----------------------------------