Add built-in var to refer to pipeline values (#3661)

crates/nu-command/src/commands/filters/collect.rs

@@ -0,0 +1,90 @@
+use crate::prelude::*;
+use nu_engine::run_block;
+use nu_engine::WholeStreamCommand;
+use nu_errors::ShellError;
+use nu_protocol::{hir::CapturedBlock, Signature, SyntaxShape, UntaggedValue};
+
+pub struct Command;
+
+impl WholeStreamCommand for Command {
+    fn name(&self) -> &str {
+        "collect"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build("collect").required(
+            "block",
+            SyntaxShape::Block,
+            "the block to run once the stream is collected",
+        )
+    }
+
+    fn usage(&self) -> &str {
+        "Collect the stream and pass it to a block."
+    }
+
+    fn run(&self, args: CommandArgs) -> Result<OutputStream, ShellError> {
+        collect(args)
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![Example {
+            description: "Use the second value in the stream",
+            example: "echo 1 2 3 | collect { |x| echo $x.1 }",
+            result: Some(vec![UntaggedValue::int(2).into()]),
+        }]
+    }
+}
+
+fn collect(args: CommandArgs) -> Result<OutputStream, ShellError> {
+    let external_redirection = args.call_info.args.external_redirection;
+    let context = &args.context;
+    let tag = args.call_info.name_tag.clone();
+    let block: CapturedBlock = args.req(0)?;
+    let mut input = args.input;
+    let param = if !block.block.params.positional.is_empty() {
+        block.block.params.positional[0].0.name()
+    } else {
+        "$it"
+    };
+
+    context.scope.enter_scope();
+
+    context.scope.add_vars(&block.captured.entries);
+    let mut input = input.drain_vec();
+    match input.len() {
+        x if x > 1 => {
+            context
+                .scope
+                .add_var(param, UntaggedValue::Table(input).into_value(tag));
+        }
+        1 => {
+            let item = input.swap_remove(0);
+            context.scope.add_var(param, item);
+        }
+        _ => {}
+    }
+
+    let result = run_block(
+        &block.block,
+        &context,
+        InputStream::empty(),
+        external_redirection,
+    );
+    context.scope.exit_scope();
+
+    Ok(result?.into_output_stream())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Command;
+    use super::ShellError;
+
+    #[test]
+    fn examples_work_as_expected() -> Result<(), ShellError> {
+        use crate::examples::test as test_examples;
+
+        test_examples(Command {})
+    }
+}

crates/nu-command/src/commands/filters/mod.rs

@@ -1,6 +1,7 @@
 mod all;
 mod any;
 mod append;
+mod collect;
 mod compact;
 mod default;
 mod drop;
@@ -41,6 +42,7 @@ mod wrap;
 pub use all::Command as All;
 pub use any::Command as Any;
 pub use append::Command as Append;
+pub use collect::Command as Collect;
 pub use compact::Compact;
 pub use default::Default;
 pub use drop::*;

crates/nu-command/src/default_context.rs

@@ -176,6 +176,7 @@ pub fn create_default_context(interactive: bool) -> Result<EvaluationContext, Bo
             whole_stream_command(RollUp),
             whole_stream_command(Rotate),
             whole_stream_command(RotateCounterClockwise),
+            whole_stream_command(Collect),
             // Data processing
             whole_stream_command(Histogram),
             whole_stream_command(Autoenv),

crates/nu-parser/src/parse.rs

@@ -10,9 +10,9 @@ use log::trace;
 use nu_errors::{ArgumentError, ParseError};
 use nu_path::{expand_path, expand_path_string};
 use nu_protocol::hir::{
-    self, Binary, Block, ClassifiedCommand, Expression, ExternalRedirection, Flag, FlagKind, Group,
+    self, Binary, Block, Call, ClassifiedCommand, Expression, ExternalRedirection, Flag, FlagKind,
-    InternalCommand, Member, NamedArguments, Operator, Pipeline, RangeOperator, SpannedExpression,
+    Group, InternalCommand, Member, NamedArguments, Operator, Pipeline, RangeOperator,
-    Unit,
+    SpannedExpression, Synthetic, Unit,
 };
 use nu_protocol::{NamedType, PositionalType, Signature, SyntaxShape, UnspannedPathMember};
 use nu_source::{HasSpan, Span, Spanned, SpannedItem};
@@ -1926,14 +1926,20 @@ fn parse_pipeline(
     let mut commands = Pipeline::new(lite_pipeline.span());
     let mut error = None;
 
-    let mut iter = lite_pipeline.commands.into_iter().peekable();
+    let pipeline_len = lite_pipeline.commands.len();
-    while let Some(lite_cmd) = iter.next() {
+    let iter = lite_pipeline.commands.into_iter().peekable();
-        let (call, err) = parse_call(lite_cmd, iter.peek().is_none(), scope);
+    for lite_cmd in iter.enumerate() {
+        let (call, err) = parse_call(lite_cmd.1, lite_cmd.0 == (pipeline_len - 1), scope);
         if error.is_none() {
             error = err;
         }
         if let Some(call) = call {
-            commands.push(call);
+            if call.has_var_usage("$in") && lite_cmd.0 > 0 {
+                let call = wrap_with_collect(call, "$in");
+                commands.push(call);
+            } else {
+                commands.push(call);
+            }
         }
     }
 
@@ -1942,6 +1948,41 @@ fn parse_pipeline(
 
 type SpannedKeyValue = (Spanned<String>, Spanned<String>);
 
+fn wrap_with_collect(call: ClassifiedCommand, var_name: &str) -> ClassifiedCommand {
+    let mut block = Block::basic();
+
+    block.block.push(Group {
+        pipelines: vec![Pipeline {
+            list: vec![call],
+            span: Span::unknown(),
+        }],
+        span: Span::unknown(),
+    });
+
+    block.params.positional = vec![(
+        PositionalType::Mandatory(var_name.into(), SyntaxShape::Any),
+        format!("implied {}", var_name),
+    )];
+
+    ClassifiedCommand::Internal(InternalCommand {
+        name: "collect".into(),
+        name_span: Span::unknown(),
+        args: Call {
+            head: Box::new(SpannedExpression {
+                expr: Expression::Synthetic(Synthetic::String("collect".into())),
+                span: Span::unknown(),
+            }),
+            positional: Some(vec![SpannedExpression {
+                expr: Expression::Block(Arc::new(block)),
+                span: Span::unknown(),
+            }]),
+            named: None,
+            span: Span::unknown(),
+            external_redirection: ExternalRedirection::Stdout,
+        },
+    })
+}
+
 fn expand_shorthand_forms(
     lite_pipeline: &LitePipeline,
 ) -> (LitePipeline, Option<SpannedKeyValue>, Option<ParseError>) {

crates/nu-path/src/lib.rs

@@ -492,7 +492,7 @@ mod tests {
                 std::matches!(expanded, Cow::Borrowed(_)),
                 "No PathBuf should be needed here (unecessary allocation)"
             );
-            assert!(&expanded == Path::new(s));
+            assert!(expanded == Path::new(s));
         }
 
         #[test]

crates/nu-protocol/src/hir.rs

@@ -42,8 +42,8 @@ impl InternalCommand {
         }
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
-        self.args.has_it_usage()
+        self.args.has_var_usage(var_name)
     }
 
     pub fn get_free_variables(&self, known_variables: &mut Vec<String>) -> Vec<String> {
@@ -85,11 +85,11 @@ pub enum ClassifiedCommand {
 }
 
 impl ClassifiedCommand {
-    fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
         match self {
-            ClassifiedCommand::Expr(expr) => expr.has_it_usage(),
+            ClassifiedCommand::Expr(expr) => expr.has_var_usage(var_name),
-            ClassifiedCommand::Dynamic(call) => call.has_it_usage(),
+            ClassifiedCommand::Dynamic(call) => call.has_var_usage(var_name),
-            ClassifiedCommand::Internal(internal) => internal.has_it_usage(),
+            ClassifiedCommand::Internal(internal) => internal.has_var_usage(var_name),
             ClassifiedCommand::Error(_) => false,
         }
     }
@@ -126,8 +126,8 @@ impl Pipeline {
         self.list.push(command);
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
-        self.list.iter().any(|cc| cc.has_it_usage())
+        self.list.iter().any(|cc| cc.has_var_usage(var_name))
     }
 }
 
@@ -152,8 +152,8 @@ impl Group {
         self.pipelines.push(pipeline);
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
-        self.pipelines.iter().any(|cc| cc.has_it_usage())
+        self.pipelines.iter().any(|cc| cc.has_var_usage(var_name))
     }
 }
 
@@ -206,13 +206,13 @@ impl Block {
         self.infer_params();
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
-        self.block.iter().any(|x| x.has_it_usage())
+        self.block.iter().any(|x| x.has_var_usage(var_name))
     }
 
     pub fn infer_params(&mut self) {
         // FIXME: re-enable inference later
-        if self.params.positional.is_empty() && self.has_it_usage() {
+        if self.params.positional.is_empty() && self.has_var_usage("$it") {
             self.params.positional = vec![(
                 PositionalType::Mandatory("$it".to_string(), SyntaxShape::Any),
                 "implied $it".to_string(),
@@ -688,8 +688,8 @@ impl SpannedExpression {
         }
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
-        self.expr.has_it_usage()
+        self.expr.has_var_usage(var_name)
     }
 
     pub fn get_free_variables(&self, known_variables: &mut Vec<String>) -> Vec<String> {
@@ -1191,24 +1191,28 @@ impl Expression {
         Expression::Boolean(b)
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
         match self {
-            Expression::Variable(name, _) if name == "$it" => true,
+            Expression::Variable(name, _) if name == var_name => true,
             Expression::Table(headers, values) => {
-                headers.iter().any(|se| se.has_it_usage())
+                headers.iter().any(|se| se.has_var_usage(var_name))
-                    || values.iter().any(|v| v.iter().any(|se| se.has_it_usage()))
+                    || values
+                        .iter()
+                        .any(|v| v.iter().any(|se| se.has_var_usage(var_name)))
             }
-            Expression::List(list) => list.iter().any(|se| se.has_it_usage()),
+            Expression::List(list) => list.iter().any(|se| se.has_var_usage(var_name)),
-            Expression::Subexpression(block) => block.has_it_usage(),
+            Expression::Subexpression(block) => block.has_var_usage(var_name),
-            Expression::Binary(binary) => binary.left.has_it_usage() || binary.right.has_it_usage(),
+            Expression::Binary(binary) => {
-            Expression::FullColumnPath(path) => path.head.has_it_usage(),
+                binary.left.has_var_usage(var_name) || binary.right.has_var_usage(var_name)
+            }
+            Expression::FullColumnPath(path) => path.head.has_var_usage(var_name),
             Expression::Range(range) => {
                 (if let Some(left) = &range.left {
-                    left.has_it_usage()
+                    left.has_var_usage(var_name)
                 } else {
                     false
                 }) || (if let Some(right) = &range.right {
-                    right.has_it_usage()
+                    right.has_var_usage(var_name)
                 } else {
                     false
                 })
@@ -1273,9 +1277,9 @@ pub enum NamedValue {
 }
 
 impl NamedValue {
-    fn has_it_usage(&self) -> bool {
+    fn has_var_usage(&self, var_name: &str) -> bool {
         if let NamedValue::Value(_, se) = self {
-            se.has_it_usage()
+            se.has_var_usage(var_name)
         } else {
             false
         }
@@ -1369,15 +1373,15 @@ impl Call {
         }
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
-        self.head.has_it_usage()
+        self.head.has_var_usage(var_name)
             || (if let Some(pos) = &self.positional {
-                pos.iter().any(|x| x.has_it_usage())
+                pos.iter().any(|x| x.has_var_usage(var_name))
             } else {
                 false
             })
             || (if let Some(named) = &self.named {
-                named.has_it_usage()
+                named.has_var_usage(var_name)
             } else {
                 false
             })
@@ -1560,8 +1564,8 @@ impl NamedArguments {
         self.named.is_empty()
     }
 
-    pub fn has_it_usage(&self) -> bool {
+    pub fn has_var_usage(&self, var_name: &str) -> bool {
-        self.iter().any(|x| x.1.has_it_usage())
+        self.iter().any(|x| x.1.has_var_usage(var_name))
     }
 
     pub fn get_free_variables(&self, known_variables: &mut Vec<String>) -> Vec<String> {

tests/shell/pipeline/commands/internal.rs

@@ -1004,6 +1004,30 @@ fn date_and_duration_overflow() {
     assert!(actual.err.contains("Duration and date addition overflow"));
 }
 
+#[test]
+fn pipeline_params_simple() {
+    let actual = nu!(
+        cwd: ".", pipeline(
+        r#"
+        echo 1 2 3 | $in.1 * $in.2
+        "#)
+    );
+
+    assert_eq!(actual.out, "6");
+}
+
+#[test]
+fn pipeline_params_inner() {
+    let actual = nu!(
+        cwd: ".", pipeline(
+        r#"
+        echo 1 2 3 | (echo $in.2 6 7 | $in.0 * $in.1 * $in.2)
+        "#)
+    );
+
+    assert_eq!(actual.out, "126");
+}
+
 mod parse {
     use nu_test_support::nu;