Table operating commands. (#1686)

* Table operating commands.

* Updated merge test for clarity.

* More clarity.

* Better like this..

crates/nu-cli/src/cli.rs

@@ -297,20 +297,25 @@ pub fn create_default_context(
             whole_stream_command(GroupBy),
             whole_stream_command(First),
             whole_stream_command(Last),
-            whole_stream_command(Skip),
             whole_stream_command(Nth),
             whole_stream_command(Drop),
             whole_stream_command(Format),
             whole_stream_command(Where),
             whole_stream_command(Compact),
             whole_stream_command(Default),
+            whole_stream_command(Skip),
+            whole_stream_command(SkipUntil),
             whole_stream_command(SkipWhile),
+            whole_stream_command(Keep),
+            whole_stream_command(KeepUntil),
+            whole_stream_command(KeepWhile),
             whole_stream_command(Range),
             whole_stream_command(Rename),
             whole_stream_command(Uniq),
             whole_stream_command(Each),
             whole_stream_command(IsEmpty),
             // Table manipulation
+            whole_stream_command(Merge),
             whole_stream_command(Shuffle),
             whole_stream_command(Wrap),
             whole_stream_command(Pivot),

crates/nu-cli/src/commands.rs

@@ -55,11 +55,15 @@ pub(crate) mod histogram;
 pub(crate) mod history;
 pub(crate) mod insert;
 pub(crate) mod is_empty;
+pub(crate) mod keep;
+pub(crate) mod keep_until;
+pub(crate) mod keep_while;
 pub(crate) mod last;
 pub(crate) mod lines;
 pub(crate) mod ls;
 #[allow(unused)]
 pub(crate) mod map_max_by;
+pub(crate) mod merge;
 pub(crate) mod mkdir;
 pub(crate) mod mv;
 pub(crate) mod next;
@@ -86,6 +90,7 @@ pub(crate) mod shells;
 pub(crate) mod shuffle;
 pub(crate) mod size;
 pub(crate) mod skip;
+pub(crate) mod skip_until;
 pub(crate) mod skip_while;
 pub(crate) mod sort_by;
 pub(crate) mod split_by;
@@ -170,11 +175,15 @@ pub(crate) use help::Help;
 pub(crate) use histogram::Histogram;
 pub(crate) use history::History;
 pub(crate) use insert::Insert;
+pub(crate) use keep::Keep;
+pub(crate) use keep_until::KeepUntil;
+pub(crate) use keep_while::KeepWhile;
 pub(crate) use last::Last;
 pub(crate) use lines::Lines;
 pub(crate) use ls::Ls;
 #[allow(unused_imports)]
 pub(crate) use map_max_by::MapMaxBy;
+pub(crate) use merge::Merge;
 pub(crate) use mkdir::Mkdir;
 pub(crate) use mv::Move;
 pub(crate) use next::Next;
@@ -199,6 +208,7 @@ pub(crate) use shells::Shells;
 pub(crate) use shuffle::Shuffle;
 pub(crate) use size::Size;
 pub(crate) use skip::Skip;
+pub(crate) use skip_until::SkipUntil;
 pub(crate) use skip_while::SkipWhile;
 pub(crate) use sort_by::SortBy;
 pub(crate) use split_by::SplitBy;

crates/nu-cli/src/commands/edit.rs

@@ -1,3 +1,4 @@
+use crate::commands::classified::block::run_block;
 use crate::commands::WholeStreamCommand;
 use crate::context::CommandRegistry;
 use crate::prelude::*;
@@ -5,6 +6,7 @@ use nu_errors::ShellError;
 use nu_protocol::{ColumnPath, ReturnSuccess, Signature, SyntaxShape, UntaggedValue, Value};
 use nu_value_ext::ValueExt;
 
+use futures::stream::once;
 pub struct Edit;
 
 #[derive(Deserialize)]
@@ -41,41 +43,104 @@ impl WholeStreamCommand for Edit {
         args: CommandArgs,
         registry: &CommandRegistry,
     ) -> Result<OutputStream, ShellError> {
-        args.process(registry, edit)?.run()
+        Ok(args.process_raw(registry, edit)?.run())
     }
 }
 
 fn edit(
     EditArgs { field, replacement }: EditArgs,
-    RunnableContext { input, .. }: RunnableContext,
+    context: RunnableContext,
+    raw_args: RawCommandArgs,
 ) -> Result<OutputStream, ShellError> {
-    let mut input = input;
+    let scope = raw_args.call_info.scope.clone();
+    let registry = context.registry.clone();
+    let mut input_stream = context.input;
 
     let stream = async_stream! {
-        match input.next().await {
-            Some(obj @ Value {
-                value: UntaggedValue::Row(_),
-                ..
-            }) => match obj.replace_data_at_column_path(&field, replacement.clone()) {
-                Some(v) => yield Ok(ReturnSuccess::Value(v)),
-                None => {
-                    yield Err(ShellError::labeled_error(
-                        "edit could not find place to insert column",
-                        "column name",
-                        obj.tag,
-                    ))
-                }
-            },
+        while let Some(input) = input_stream.next().await {
+        let replacement = replacement.clone();
+        match replacement {
+            Value {
+                value: UntaggedValue::Block(block),
+                tag,
+            } =>  {
+                let mut context = Context::from_raw(&raw_args, &registry);
+                let for_block = input.clone();
+                let input_clone = input.clone();
+                let input_stream = once(async { Ok(for_block) }).to_input_stream();
 
-            Some(Value { tag, ..}) => {
-                yield Err(ShellError::labeled_error(
-                    "Unrecognized type in stream",
-                    "original value",
-                    tag,
-                ))
+                let result = run_block(
+                    &block,
+                    &mut context,
+                    input_stream,
+                    &scope.clone().set_it(input_clone),
+                ).await;
+
+                match result {
+                    Ok(mut stream) => {
+                        let errors = context.get_errors();
+                        if let Some(error) = errors.first() {
+                            yield Err(error.clone());
+                        }
+
+                        match input {
+                            obj @ Value {
+                                value: UntaggedValue::Row(_),
+                                ..
+                            } => {
+                                if let Some(result) = stream.next().await {
+                                    match obj.replace_data_at_column_path(&field, result.clone()) {
+                                        Some(v) => yield Ok(ReturnSuccess::Value(v)),
+                                        None => {
+                                            yield Err(ShellError::labeled_error(
+                                                "edit could not find place to insert column",
+                                                "column name",
+                                                obj.tag,
+                                            ))
+                                        }
+                                    }
+                                }
+                            }
+                            Value { tag, ..} => {
+                                yield Err(ShellError::labeled_error(
+                                    "Unrecognized type in stream",
+                                    "original value",
+                                    tag,
+                                ))
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        yield Err(e);
+                    }
+                }
             }
-            _ => {}
-        }
+            _ => {
+                match input {
+                    obj @ Value {
+                        value: UntaggedValue::Row(_),
+                        ..
+                    } => match obj.replace_data_at_column_path(&field, replacement.clone()) {
+                        Some(v) => yield Ok(ReturnSuccess::Value(v)),
+                        None => {
+                            yield Err(ShellError::labeled_error(
+                                "edit could not find place to insert column",
+                                "column name",
+                                obj.tag,
+                            ))
+                        }
+                    },
+                    Value { tag, ..} => {
+                        yield Err(ShellError::labeled_error(
+                            "Unrecognized type in stream",
+                            "original value",
+                            tag,
+                        ))
+                    }
+                    _ => {}
+                }
+            }
+        }}
     };
 
     Ok(stream.to_output_stream())

crates/nu-cli/src/commands/keep.rs

@@ -0,0 +1,49 @@
+use crate::commands::WholeStreamCommand;
+use crate::context::CommandRegistry;
+use crate::prelude::*;
+use nu_errors::ShellError;
+use nu_protocol::{Signature, SyntaxShape};
+use nu_source::Tagged;
+
+pub struct Keep;
+
+#[derive(Deserialize)]
+pub struct KeepArgs {
+    rows: Option<Tagged<usize>>,
+}
+
+impl WholeStreamCommand for Keep {
+    fn name(&self) -> &str {
+        "keep"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("keep").optional(
+            "rows",
+            SyntaxShape::Int,
+            "starting from the front, the number of rows to keep",
+        )
+    }
+
+    fn usage(&self) -> &str {
+        "Keep the number of rows only"
+    }
+
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        args.process(registry, keep)?.run()
+    }
+}
+
+fn keep(KeepArgs { rows }: KeepArgs, context: RunnableContext) -> Result<OutputStream, ShellError> {
+    let rows_desired = if let Some(quantity) = rows {
+        *quantity
+    } else {
+        1
+    };
+
+    Ok(OutputStream::from_input(context.input.take(rows_desired)))
+}

crates/nu-cli/src/commands/keep_until.rs

@@ -0,0 +1,98 @@
+use crate::commands::WholeStreamCommand;
+use crate::evaluate::evaluate_baseline_expr;
+use crate::prelude::*;
+use log::trace;
+use nu_errors::ShellError;
+use nu_protocol::{hir::ClassifiedCommand, Signature, SyntaxShape, UntaggedValue, Value};
+
+pub struct KeepUntil;
+
+impl WholeStreamCommand for KeepUntil {
+    fn name(&self) -> &str {
+        "keep-until"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("keep-until")
+            .required(
+                "condition",
+                SyntaxShape::Math,
+                "the condition that must be met to stop keeping rows",
+            )
+            .filter()
+    }
+
+    fn usage(&self) -> &str {
+        "Keeps rows until the condition matches."
+    }
+
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        let registry = registry.clone();
+        let scope = args.call_info.scope.clone();
+        let call_info = args.evaluate_once(&registry)?;
+
+        let block = call_info.args.expect_nth(0)?.clone();
+
+        let condition = match block {
+            Value {
+                value: UntaggedValue::Block(block),
+                tag,
+            } => {
+                if block.block.len() != 1 {
+                    return Err(ShellError::labeled_error(
+                        "Expected a condition",
+                        "expected a condition",
+                        tag,
+                    ));
+                }
+                match block.block[0].list.get(0) {
+                    Some(item) => match item {
+                        ClassifiedCommand::Expr(expr) => expr.clone(),
+                        _ => {
+                            return Err(ShellError::labeled_error(
+                                "Expected a condition",
+                                "expected a condition",
+                                tag,
+                            ))
+                        }
+                    },
+                    None => {
+                        return Err(ShellError::labeled_error(
+                            "Expected a condition",
+                            "expected a condition",
+                            tag,
+                        ));
+                    }
+                }
+            }
+            Value { tag, .. } => {
+                return Err(ShellError::labeled_error(
+                    "Expected a condition",
+                    "expected a condition",
+                    tag,
+                ));
+            }
+        };
+
+        let objects = call_info.input.take_while(move |item| {
+            let condition = condition.clone();
+            trace!("ITEM = {:?}", item);
+            let result =
+                evaluate_baseline_expr(&*condition, &registry, &scope.clone().set_it(item.clone()));
+            trace!("RESULT = {:?}", result);
+
+            let return_value = match result {
+                Ok(ref v) if v.is_true() => false,
+                _ => true,
+            };
+
+            futures::future::ready(return_value)
+        });
+
+        Ok(objects.from_input_stream())
+    }
+}

crates/nu-cli/src/commands/keep_while.rs

@@ -0,0 +1,98 @@
+use crate::commands::WholeStreamCommand;
+use crate::evaluate::evaluate_baseline_expr;
+use crate::prelude::*;
+use log::trace;
+use nu_errors::ShellError;
+use nu_protocol::{hir::ClassifiedCommand, Signature, SyntaxShape, UntaggedValue, Value};
+
+pub struct KeepWhile;
+
+impl WholeStreamCommand for KeepWhile {
+    fn name(&self) -> &str {
+        "keep-while"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("keep-while")
+            .required(
+                "condition",
+                SyntaxShape::Math,
+                "the condition that must be met to keep rows",
+            )
+            .filter()
+    }
+
+    fn usage(&self) -> &str {
+        "Keeps rows while the condition matches."
+    }
+
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        let registry = registry.clone();
+        let scope = args.call_info.scope.clone();
+        let call_info = args.evaluate_once(&registry)?;
+
+        let block = call_info.args.expect_nth(0)?.clone();
+
+        let condition = match block {
+            Value {
+                value: UntaggedValue::Block(block),
+                tag,
+            } => {
+                if block.block.len() != 1 {
+                    return Err(ShellError::labeled_error(
+                        "Expected a condition",
+                        "expected a condition",
+                        tag,
+                    ));
+                }
+                match block.block[0].list.get(0) {
+                    Some(item) => match item {
+                        ClassifiedCommand::Expr(expr) => expr.clone(),
+                        _ => {
+                            return Err(ShellError::labeled_error(
+                                "Expected a condition",
+                                "expected a condition",
+                                tag,
+                            ))
+                        }
+                    },
+                    None => {
+                        return Err(ShellError::labeled_error(
+                            "Expected a condition",
+                            "expected a condition",
+                            tag,
+                        ));
+                    }
+                }
+            }
+            Value { tag, .. } => {
+                return Err(ShellError::labeled_error(
+                    "Expected a condition",
+                    "expected a condition",
+                    tag,
+                ));
+            }
+        };
+
+        let objects = call_info.input.take_while(move |item| {
+            let condition = condition.clone();
+            trace!("ITEM = {:?}", item);
+            let result =
+                evaluate_baseline_expr(&*condition, &registry, &scope.clone().set_it(item.clone()));
+            trace!("RESULT = {:?}", result);
+
+            let return_value = match result {
+                Ok(ref v) if v.is_true() => true,
+                _ => false,
+            };
+
+            futures::future::ready(return_value)
+        });
+
+        Ok(objects.from_input_stream())
+    }
+}

crates/nu-cli/src/commands/merge.rs

@@ -0,0 +1,95 @@
+use crate::commands::classified::block::run_block;
+use crate::commands::WholeStreamCommand;
+use crate::context::CommandRegistry;
+use crate::data::value::merge_values;
+use crate::prelude::*;
+
+use indexmap::IndexMap;
+use nu_errors::ShellError;
+use nu_protocol::{hir::Block, ReturnSuccess, Scope, Signature, SyntaxShape, UntaggedValue, Value};
+pub struct Merge;
+
+#[derive(Deserialize)]
+pub struct MergeArgs {
+    block: Block,
+}
+
+impl WholeStreamCommand for Merge {
+    fn name(&self) -> &str {
+        "merge"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("merge").required(
+            "block",
+            SyntaxShape::Block,
+            "the block to run and merge into the table",
+        )
+    }
+
+    fn usage(&self) -> &str {
+        "Merge a table."
+    }
+
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        Ok(args.process_raw(registry, merge)?.run())
+    }
+}
+
+fn merge(
+    merge_args: MergeArgs,
+    context: RunnableContext,
+    raw_args: RawCommandArgs,
+) -> Result<OutputStream, ShellError> {
+    let block = merge_args.block;
+    let registry = context.registry.clone();
+    let mut input = context.input;
+
+    let mut context = Context::from_raw(&raw_args, &registry);
+
+    let stream = async_stream! {
+        let table: Option<Vec<Value>> = match run_block(&block,
+                &mut context,
+                InputStream::empty(),
+                &Scope::empty()).await {
+            Ok(mut stream) => Some(stream.drain_vec().await),
+            Err(err) => {
+                yield Err(err);
+                return;
+            }
+        };
+
+
+        let table = table.unwrap_or_else(|| vec![Value {
+            value: UntaggedValue::row(IndexMap::default()),
+            tag: raw_args.call_info.name_tag,
+        }]);
+
+        let mut idx = 0;
+
+        while let Some(value) = input.next().await {
+            let other = table.get(idx);
+
+            match other {
+                Some(replacement) => {
+                    match merge_values(&value.value, &replacement.value) {
+                        Ok(merged_value) => yield ReturnSuccess::value(merged_value.into_value(&value.tag)),
+                        Err(err) => {
+                            let message = format!("The row at {:?} types mismatch", idx);
+                            yield Err(ShellError::labeled_error("Could not merge", &message, &value.tag));
+                        }
+                    }
+                }
+                None => yield ReturnSuccess::value(value),
+            }
+
+            idx += 1;
+        }
+    };
+
+    Ok(stream.to_output_stream())
+}

crates/nu-cli/src/commands/skip_until.rs

@@ -0,0 +1,98 @@
+use crate::commands::WholeStreamCommand;
+use crate::evaluate::evaluate_baseline_expr;
+use crate::prelude::*;
+use log::trace;
+use nu_errors::ShellError;
+use nu_protocol::{hir::ClassifiedCommand, Signature, SyntaxShape, UntaggedValue, Value};
+
+pub struct SkipUntil;
+
+impl WholeStreamCommand for SkipUntil {
+    fn name(&self) -> &str {
+        "skip-until"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("skip-until")
+            .required(
+                "condition",
+                SyntaxShape::Math,
+                "the condition that must be met to stop skipping",
+            )
+            .filter()
+    }
+
+    fn usage(&self) -> &str {
+        "Skips rows until the condition matches."
+    }
+
+    fn run(
+        &self,
+        args: CommandArgs,
+        registry: &CommandRegistry,
+    ) -> Result<OutputStream, ShellError> {
+        let registry = registry.clone();
+        let scope = args.call_info.scope.clone();
+        let call_info = args.evaluate_once(&registry)?;
+
+        let block = call_info.args.expect_nth(0)?.clone();
+
+        let condition = match block {
+            Value {
+                value: UntaggedValue::Block(block),
+                tag,
+            } => {
+                if block.block.len() != 1 {
+                    return Err(ShellError::labeled_error(
+                        "Expected a condition",
+                        "expected a condition",
+                        tag,
+                    ));
+                }
+                match block.block[0].list.get(0) {
+                    Some(item) => match item {
+                        ClassifiedCommand::Expr(expr) => expr.clone(),
+                        _ => {
+                            return Err(ShellError::labeled_error(
+                                "Expected a condition",
+                                "expected a condition",
+                                tag,
+                            ))
+                        }
+                    },
+                    None => {
+                        return Err(ShellError::labeled_error(
+                            "Expected a condition",
+                            "expected a condition",
+                            tag,
+                        ));
+                    }
+                }
+            }
+            Value { tag, .. } => {
+                return Err(ShellError::labeled_error(
+                    "Expected a condition",
+                    "expected a condition",
+                    tag,
+                ));
+            }
+        };
+
+        let objects = call_info.input.skip_while(move |item| {
+            let condition = condition.clone();
+            trace!("ITEM = {:?}", item);
+            let result =
+                evaluate_baseline_expr(&*condition, &registry, &scope.clone().set_it(item.clone()));
+            trace!("RESULT = {:?}", result);
+
+            let return_value = match result {
+                Ok(ref v) if v.is_true() => false,
+                _ => true,
+            };
+
+            futures::future::ready(return_value)
+        });
+
+        Ok(objects.from_input_stream())
+    }
+}

crates/nu-cli/src/data/value.rs

@@ -22,6 +22,18 @@ pub fn date_from_str(s: Tagged<&str>) -> Result<UntaggedValue, ShellError> {
     Ok(UntaggedValue::Primitive(Primitive::Date(date)))
 }
 
+pub fn merge_values(
+    left: &UntaggedValue,
+    right: &UntaggedValue,
+) -> Result<UntaggedValue, (&'static str, &'static str)> {
+    match (left, right) {
+        (UntaggedValue::Row(columns), UntaggedValue::Row(columns_b)) => {
+            Ok(UntaggedValue::Row(columns.merge_from(columns_b)))
+        }
+        (left, right) => Err((left.type_name(), right.type_name())),
+    }
+}
+
 pub fn compute_values(
     operator: Operator,
     left: &UntaggedValue,
@@ -153,3 +165,31 @@ pub fn format_for_column<'a>(
         .format_for_column(column)
         .pretty()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::UntaggedValue as v;
+    use indexmap::indexmap;
+
+    use super::merge_values;
+
+    #[test]
+    fn merges_tables() {
+        let table_author_row = v::row(indexmap! {
+            "name".into() => v::string("Andrés").into_untagged_value(),
+            "country".into() => v::string("EC").into_untagged_value(),
+            "date".into() => v::string("April 29-2020").into_untagged_value()
+        });
+
+        let other_table_author_row = v::row(indexmap! {
+            "name".into() => v::string("YK").into_untagged_value(),
+            "country".into() => v::string("US").into_untagged_value(),
+            "date".into() => v::string("October 10-2019").into_untagged_value()
+        });
+
+        assert_eq!(
+            other_table_author_row,
+            merge_values(&table_author_row, &other_table_author_row).unwrap()
+        );
+    }
+}