Reductions placeholder.

This commit is contained in:
Andrés N. Robalino 2019-11-03 20:55:34 -05:00
parent 11e4410d1c
commit 21f48577ae
3 changed files with 121 additions and 6 deletions

View file

@ -323,10 +323,17 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
whole_stream_command(Table), whole_stream_command(Table),
whole_stream_command(Version), whole_stream_command(Version),
whole_stream_command(Which), whole_stream_command(Which),
#[cfg(data_processing_primitives)]
whole_stream_command(SplitBy),
]); ]);
cfg_if::cfg_if! {
if #[cfg(data_processing_primitives)] {
context.add_commands(vec![
whole_stream_command(SplitBy),
whole_stream_command(ReduceBy),
]);
}
}
#[cfg(feature = "clipboard")] #[cfg(feature = "clipboard")]
{ {
context.add_commands(vec![whole_stream_command( context.add_commands(vec![whole_stream_command(

View file

@ -58,8 +58,12 @@ pub(crate) mod size;
pub(crate) mod skip_while; pub(crate) mod skip_while;
pub(crate) mod sort_by; pub(crate) mod sort_by;
#[cfg(data_processing_primitives)] cfg_if::cfg_if! {
if #[cfg(data_processing_primitives)] {
pub(crate) mod split_by; pub(crate) mod split_by;
pub(crate) mod reduce_by;
}
}
pub(crate) mod split_column; pub(crate) mod split_column;
pub(crate) mod split_row; pub(crate) mod split_row;
@ -138,8 +142,12 @@ pub(crate) use size::Size;
pub(crate) use skip_while::SkipWhile; pub(crate) use skip_while::SkipWhile;
pub(crate) use sort_by::SortBy; pub(crate) use sort_by::SortBy;
#[cfg(data_processing_primitives)] cfg_if::cfg_if! {
if #[cfg(data_processing_primitives)] {
pub(crate) use split_by::SplitBy; pub(crate) use split_by::SplitBy;
pub(crate) use reduce_by::ReduceBy;
}
}
pub(crate) use split_column::SplitColumn; pub(crate) use split_column::SplitColumn;
pub(crate) use split_row::SplitRow; pub(crate) use split_row::SplitRow;

100
src/commands/reduce_by.rs Normal file
View file

@ -0,0 +1,100 @@
use crate::commands::WholeStreamCommand;
use crate::data::TaggedDictBuilder;
use crate::parser::hir::SyntaxShape;
use crate::parser::registry;
use crate::data::base::Block;
use crate::prelude::*;
use log::trace;
pub struct ReduceBy;
#[derive(Deserialize)]
pub struct ReduceByArgs {
calculator: Block,
}
impl WholeStreamCommand for ReduceBy {
fn name(&self) -> &str {
"reduce-by"
}
fn signature(&self) -> Signature {
Signature::build("reduce-by").required(
"calculator",
SyntaxShape::Block,
"The block used for calculating values",
)
}
fn usage(&self) -> &str {
"Crates a new table with the data from the table rows reduced by the block given."
}
fn run(
&self,
args: CommandArgs,
registry: &CommandRegistry,
) -> Result<OutputStream, ShellError> {
args.process(registry, reduce_by)?.run()
}
}
pub fn reduce_by(
ReduceByArgs { calculator }: ReduceByArgs,
RunnableContext { input, name, .. }: RunnableContext,
) -> Result<OutputStream, ShellError> {
let stream = async_stream! {
let values: Vec<Tagged<Value>> = input.values.collect().await;
trace!("{:?}", &calculator);
if values.is_empty() {
yield Err(ShellError::labeled_error(
"Expected table from pipeline",
"requires a table input",
name
))
} else {
match reduce(values, &calculator, name) {
Ok(reduced) => yield ReturnSuccess::value(reduced),
Err(err) => yield Err(err)
}
}
};
Ok(stream.to_output_stream())
}
pub fn reduce(
values: Vec<Tagged<Value>>,
calculator: &Block,
tag: impl Into<Tag>,
) -> Result<Tagged<Value>, ShellError> {
let tag = tag.into();
let mut out = TaggedDictBuilder::new(&tag);
Ok(out.into_tagged_value())
}
#[cfg(test)]
mod tests {
use crate::commands::reduce_by::reduce;
use crate::data::meta::*;
use crate::Value;
use indexmap::IndexMap;
fn string(input: impl Into<String>) -> Tagged<Value> {
Value::string(input.into()).tagged_unknown()
}
fn row(entries: IndexMap<String, Tagged<Value>>) -> Tagged<Value> {
Value::row(entries).tagged_unknown()
}
fn table(list: &Vec<Tagged<Value>>) -> Tagged<Value> {
Value::table(list).tagged_unknown()
}
}