Improve range and internal iteration (#3300)

crates/nu-cli/src/prelude.rs

@@ -8,19 +8,6 @@ macro_rules! return_err {
     };
 }
 
-#[macro_export]
-macro_rules! stream {
-    ($($expr:expr),*) => {{
-        let mut v = VecDeque::new();
-
-        $(
-            v.push_back($expr);
-        )*
-
-        v
-    }}
-}
-
 #[macro_export]
 macro_rules! trace_out_stream {
     (target: $target:tt, $desc:tt = $expr:expr) => {{

crates/nu-command/src/commands/echo.rs

@@ -1,4 +1,5 @@
 use crate::prelude::*;
+use bigdecimal::Zero;
 use nu_engine::WholeStreamCommand;
 use nu_errors::ShellError;
 use nu_protocol::hir::Operator;
@@ -69,11 +70,13 @@ fn echo(args: CommandArgs) -> Result<OutputStream, ShellError> {
 }
 
 struct RangeIterator {
-    curr: Primitive,
+    curr: UntaggedValue,
-    end: Primitive,
+    end: UntaggedValue,
     tag: Tag,
     is_end_inclusive: bool,
     moves_up: bool,
+    one: UntaggedValue,
+    negative_one: UntaggedValue,
 }
 
 impl RangeIterator {
@@ -90,10 +93,12 @@ impl RangeIterator {
 
         RangeIterator {
             moves_up: start <= end,
-            curr: start,
+            curr: UntaggedValue::Primitive(start),
-            end,
+            end: UntaggedValue::Primitive(end),
             tag,
             is_end_inclusive: matches!(range.to.1, RangeInclusion::Inclusive),
+            one: UntaggedValue::int(1),
+            negative_one: UntaggedValue::int(-1),
         }
     }
 }
@@ -101,50 +106,46 @@ impl RangeIterator {
 impl Iterator for RangeIterator {
     type Item = Result<ReturnSuccess, ShellError>;
     fn next(&mut self) -> Option<Self::Item> {
-        let ordering = if self.end == Primitive::Nothing {
+        use std::cmp::Ordering;
+
+        let ordering = if self.end == UntaggedValue::Primitive(Primitive::Nothing) {
             Ordering::Less
         } else {
-            let result =
+            match (&self.curr, &self.end) {
-                nu_data::base::coerce_compare_primitive(&self.curr, &self.end).map_err(|_| {
+                (
-                    ShellError::labeled_error(
+                    UntaggedValue::Primitive(Primitive::Int(x)),
+                    UntaggedValue::Primitive(Primitive::Int(y)),
+                ) => x.cmp(y),
+                (
+                    UntaggedValue::Primitive(Primitive::Decimal(x)),
+                    UntaggedValue::Primitive(Primitive::Decimal(y)),
+                ) => x.cmp(y),
+                (
+                    UntaggedValue::Primitive(Primitive::Decimal(x)),
+                    UntaggedValue::Primitive(Primitive::Int(y)),
+                ) => x.cmp(&(BigDecimal::zero() + y)),
+                (
+                    UntaggedValue::Primitive(Primitive::Int(x)),
+                    UntaggedValue::Primitive(Primitive::Decimal(y)),
+                ) => (BigDecimal::zero() + x).cmp(y),
+                _ => {
+                    return Some(Err(ShellError::labeled_error(
                         "Cannot create range",
                         "unsupported range",
                         self.tag.span,
-                    )
+                    )))
-                });
+                }
-
-            if let Err(result) = result {
-                return Some(Err(result));
             }
-
-            let result = result
-                .expect("Internal error: the error case was already protected, but that failed");
-
-            result.compare()
         };
 
-        use std::cmp::Ordering;
-
         if self.moves_up
             && (ordering == Ordering::Less || self.is_end_inclusive && ordering == Ordering::Equal)
         {
-            let output = UntaggedValue::Primitive(self.curr.clone()).into_value(self.tag.clone());
+            let next_value = nu_data::value::compute_values(Operator::Plus, &self.curr, &self.one);
 
-            let next_value = nu_data::value::compute_values(
+            let mut next = match next_value {
-                Operator::Plus,
+                Ok(result) => result,
-                &UntaggedValue::Primitive(self.curr.clone()),
-                &UntaggedValue::int(1),
-            );
 
-            self.curr = match next_value {
-                Ok(result) => match result {
-                    UntaggedValue::Primitive(p) => p,
-                    _ => {
-                        return Some(Err(ShellError::unimplemented(
-                            "Internal error: expected a primitive result from increment",
-                        )));
-                    }
-                },
                 Err((left_type, right_type)) => {
                     return Some(Err(ShellError::coerce_error(
                         left_type.spanned(self.tag.span),
@@ -152,28 +153,18 @@ impl Iterator for RangeIterator {
                     )));
                 }
             };
-            Some(ReturnSuccess::value(output))
+            std::mem::swap(&mut self.curr, &mut next);
+
+            Some(ReturnSuccess::value(next.into_value(self.tag.clone())))
         } else if !self.moves_up
             && (ordering == Ordering::Greater
                 || self.is_end_inclusive && ordering == Ordering::Equal)
         {
-            let output = UntaggedValue::Primitive(self.curr.clone()).into_value(self.tag.clone());
+            let next_value =
+                nu_data::value::compute_values(Operator::Plus, &self.curr, &self.negative_one);
 
-            let next_value = nu_data::value::compute_values(
+            let mut next = match next_value {
-                Operator::Plus,
+                Ok(result) => result,
-                &UntaggedValue::Primitive(self.curr.clone()),
-                &UntaggedValue::int(-1),
-            );
-
-            self.curr = match next_value {
-                Ok(result) => match result {
-                    UntaggedValue::Primitive(p) => p,
-                    _ => {
-                        return Some(Err(ShellError::unimplemented(
-                            "Internal error: expected a primitive result from increment",
-                        )));
-                    }
-                },
                 Err((left_type, right_type)) => {
                     return Some(Err(ShellError::coerce_error(
                         left_type.spanned(self.tag.span),
@@ -181,7 +172,9 @@ impl Iterator for RangeIterator {
                     )));
                 }
             };
-            Some(ReturnSuccess::value(output))
+            std::mem::swap(&mut self.curr, &mut next);
+
+            Some(ReturnSuccess::value(next.into_value(self.tag.clone())))
         } else {
             None
         }

crates/nu-command/src/prelude.rs

@@ -8,19 +8,6 @@ macro_rules! return_err {
     };
 }
 
-#[macro_export]
-macro_rules! stream {
-    ($($expr:expr),*) => {{
-        let mut v = VecDeque::new();
-
-        $(
-            v.push_back($expr);
-        )*
-
-        v
-    }}
-}
-
 pub(crate) use bigdecimal::BigDecimal;
 pub(crate) use indexmap::{indexmap, IndexMap};
 pub(crate) use itertools::Itertools;

crates/nu-engine/src/command_args.rs

@@ -33,16 +33,33 @@ pub struct CommandArgs {
 
 pub type RunnableContext = CommandArgs;
 
+#[derive(Clone)]
 pub struct RunnableContextWithoutInput {
     pub shell_manager: ShellManager,
     pub host: Arc<parking_lot::Mutex<Box<dyn Host>>>,
     pub current_errors: Arc<Mutex<Vec<ShellError>>>,
     pub ctrl_c: Arc<AtomicBool>,
+    pub call_info: UnevaluatedCallInfo,
     pub configs: Arc<Mutex<ConfigHolder>>,
     pub scope: Scope,
     pub name: Tag,
 }
 
+impl RunnableContextWithoutInput {
+    pub fn with_input(self, input: InputStream) -> CommandArgs {
+        CommandArgs {
+            shell_manager: self.shell_manager,
+            host: self.host,
+            current_errors: self.current_errors,
+            ctrl_c: self.ctrl_c,
+            call_info: self.call_info,
+            configs: self.configs,
+            scope: self.scope,
+            input,
+        }
+    }
+}
+
 #[derive(Getters, Clone)]
 #[get = "pub"]
 pub struct RawCommandArgs {
@@ -108,9 +125,10 @@ impl CommandArgs {
             host: self.host,
             ctrl_c: self.ctrl_c,
             configs: self.configs,
+            name: self.call_info.name_tag.clone(),
+            call_info: self.call_info,
             current_errors: self.current_errors,
             scope: self.scope,
-            name: self.call_info.name_tag,
         };
 
         (self.input, new_context)

crates/nu-engine/src/evaluate/internal.rs

@@ -8,8 +8,7 @@ use nu_errors::ShellError;
 use nu_protocol::hir::{ExternalRedirection, InternalCommand};
 use nu_protocol::{CommandAction, ReturnSuccess, UntaggedValue, Value};
 use nu_source::{PrettyDebug, Span, Tag};
-use nu_stream::{InputStream, ToInputStream};
+use nu_stream::{InputStream, OutputStream};
-use std::sync::Arc;
 
 pub(crate) fn run_internal_command(
     command: InternalCommand,
@@ -34,194 +33,202 @@ pub(crate) fn run_internal_command(
         )?
     };
 
-    let head = Arc::new(command.args.head.clone());
-    let context = context.clone();
-    let command = Arc::new(command);
-
     Ok(InputStream::from_stream(
-        result
+        InternalIterator {
-            .map(move |item| {
+            command,
-                let head = head.clone();
+            context: context.clone(),
-                let command = command.clone();
+            leftovers: vec![],
-                let context = context.clone();
+            input: result,
-                match item {
+        }
-                    Ok(ReturnSuccess::Action(action)) => match action {
+        .take_while(|x| !x.is_error()),
-                        CommandAction::ChangePath(path) => {
-                            context.shell_manager.set_path(path);
-                            InputStream::empty()
-                        }
-                        CommandAction::Exit(code) => std::process::exit(code), // TODO: save history.txt
-                        CommandAction::Error(err) => {
-                            context.error(err);
-                            InputStream::empty()
-                        }
-                        CommandAction::AutoConvert(tagged_contents, extension) => {
-                            let contents_tag = tagged_contents.tag.clone();
-                            let command_name = format!("from {}", extension);
-                            let command = command;
-                            if let Some(converter) = context.scope.get_command(&command_name) {
-                                let new_args = RawCommandArgs {
-                                    host: context.host.clone(),
-                                    ctrl_c: context.ctrl_c.clone(),
-                                    configs: context.configs.clone(),
-                                    current_errors: context.current_errors.clone(),
-                                    shell_manager: context.shell_manager.clone(),
-                                    call_info: UnevaluatedCallInfo {
-                                        args: nu_protocol::hir::Call {
-                                            head: (&*head).clone(),
-                                            positional: None,
-                                            named: None,
-                                            span: Span::unknown(),
-                                            external_redirection: ExternalRedirection::Stdout,
-                                        },
-                                        name_tag: Tag::unknown_anchor(command.name_span),
-                                    },
-                                    scope: context.scope.clone(),
-                                };
-                                let result =
-                                    converter.run(new_args.with_input(vec![tagged_contents]));
-
-                                match result {
-                                    Ok(mut result) => {
-                                        let result_vec: Vec<Result<ReturnSuccess, ShellError>> =
-                                            result.drain_vec();
-
-                                        let mut output = vec![];
-                                        for res in result_vec {
-                                            match res {
-                                                Ok(ReturnSuccess::Value(Value {
-                                                    value: UntaggedValue::Table(list),
-                                                    ..
-                                                })) => {
-                                                    for l in list {
-                                                        output.push(Ok(l));
-                                                    }
-                                                }
-                                                Ok(ReturnSuccess::Value(Value {
-                                                    value, ..
-                                                })) => {
-                                                    output.push(Ok(
-                                                        value.into_value(contents_tag.clone())
-                                                    ));
-                                                }
-                                                Err(e) => output.push(Err(e)),
-                                                _ => {}
-                                            }
-                                        }
-
-                                        output.into_iter().to_input_stream()
-                                    }
-                                    Err(err) => {
-                                        context.error(err);
-                                        InputStream::empty()
-                                    }
-                                }
-                            } else {
-                                InputStream::one(tagged_contents)
-                            }
-                        }
-                        CommandAction::EnterValueShell(value) => {
-                            context
-                                .shell_manager
-                                .insert_at_current(Box::new(ValueShell::new(value)));
-                            InputStream::from_stream(std::iter::empty())
-                        }
-                        CommandAction::EnterShell(location) => {
-                            let mode = if context.shell_manager.is_interactive() {
-                                FilesystemShellMode::Cli
-                            } else {
-                                FilesystemShellMode::Script
-                            };
-                            context.shell_manager.insert_at_current(Box::new(
-                                match FilesystemShell::with_location(location, mode) {
-                                    Ok(v) => v,
-                                    Err(err) => {
-                                        context.error(err.into());
-                                        return InputStream::empty();
-                                    }
-                                },
-                            ));
-                            InputStream::from_stream(std::iter::empty())
-                        }
-                        CommandAction::AddPlugins(path) => {
-                            match crate::plugin::build_plugin::scan(vec![std::path::PathBuf::from(
-                                path,
-                            )]) {
-                                Ok(plugins) => {
-                                    context.add_commands(
-                                        plugins
-                                            .into_iter()
-                                            .filter(|p| !context.is_command_registered(p.name()))
-                                            .collect(),
-                                    );
-
-                                    InputStream::empty()
-                                }
-                                Err(reason) => {
-                                    context.error(reason);
-                                    InputStream::empty()
-                                }
-                            }
-                        }
-                        CommandAction::PreviousShell => {
-                            context.shell_manager.prev();
-                            InputStream::empty()
-                        }
-                        CommandAction::NextShell => {
-                            context.shell_manager.next();
-                            InputStream::empty()
-                        }
-                        CommandAction::LeaveShell(code) => {
-                            context.shell_manager.remove_at_current();
-                            if context.shell_manager.is_empty() {
-                                std::process::exit(code); // TODO: save history.txt
-                            }
-                            InputStream::empty()
-                        }
-                        CommandAction::UnloadConfig(cfg_path) => {
-                            context.unload_config(&cfg_path);
-                            InputStream::empty()
-                        }
-                        CommandAction::LoadConfig(cfg_path) => {
-                            if let Err(e) = context.load_config(&cfg_path) {
-                                InputStream::one(UntaggedValue::Error(e).into_untagged_value())
-                            } else {
-                                InputStream::empty()
-                            }
-                        }
-                    },
-
-                    Ok(ReturnSuccess::Value(Value {
-                        value: UntaggedValue::Error(err),
-                        ..
-                    })) => {
-                        context.error(err);
-                        InputStream::empty()
-                    }
-
-                    Ok(ReturnSuccess::Value(v)) => InputStream::one(v),
-
-                    Ok(ReturnSuccess::DebugValue(v)) => {
-                        let doc = PrettyDebug::pretty_doc(&v);
-                        let mut buffer = termcolor::Buffer::ansi();
-
-                        let _ = doc.render_raw(
-                            context.with_host(|host| host.width() - 5),
-                            &mut nu_source::TermColored::new(&mut buffer),
-                        );
-
-                        let value = String::from_utf8_lossy(buffer.as_slice());
-
-                        InputStream::one(UntaggedValue::string(value).into_untagged_value())
-                    }
-
-                    Err(err) => {
-                        context.error(err);
-                        InputStream::empty()
-                    }
-                }
-            })
-            .flatten()
-            .take_while(|x| !x.is_error()),
     ))
 }
+
+struct InternalIterator {
+    context: EvaluationContext,
+    command: InternalCommand,
+    leftovers: Vec<Value>,
+    input: OutputStream,
+}
+
+impl Iterator for InternalIterator {
+    type Item = Value;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        // let head = Arc::new(command.args.head.clone());
+        // let context = context.clone();
+        // let command = Arc::new(command);
+
+        if !self.leftovers.is_empty() {
+            let output = self.leftovers.remove(0);
+            return Some(output);
+        }
+
+        while let Some(item) = self.input.next() {
+            match item {
+                Ok(ReturnSuccess::Action(action)) => match action {
+                    CommandAction::ChangePath(path) => {
+                        self.context.shell_manager.set_path(path);
+                    }
+                    CommandAction::Exit(code) => std::process::exit(code), // TODO: save history.txt
+                    CommandAction::Error(err) => {
+                        self.context.error(err);
+                    }
+                    CommandAction::AutoConvert(tagged_contents, extension) => {
+                        let contents_tag = tagged_contents.tag.clone();
+                        let command_name = format!("from {}", extension);
+                        if let Some(converter) = self.context.scope.get_command(&command_name) {
+                            let new_args = RawCommandArgs {
+                                host: self.context.host.clone(),
+                                ctrl_c: self.context.ctrl_c.clone(),
+                                configs: self.context.configs.clone(),
+                                current_errors: self.context.current_errors.clone(),
+                                shell_manager: self.context.shell_manager.clone(),
+                                call_info: UnevaluatedCallInfo {
+                                    args: nu_protocol::hir::Call {
+                                        head: self.command.args.head.clone(),
+                                        positional: None,
+                                        named: None,
+                                        span: Span::unknown(),
+                                        external_redirection: ExternalRedirection::Stdout,
+                                    },
+                                    name_tag: Tag::unknown_anchor(self.command.name_span),
+                                },
+                                scope: self.context.scope.clone(),
+                            };
+                            let result = converter.run(new_args.with_input(vec![tagged_contents]));
+
+                            match result {
+                                Ok(mut result) => {
+                                    let result_vec: Vec<Result<ReturnSuccess, ShellError>> =
+                                        result.drain_vec();
+
+                                    let mut output = vec![];
+                                    for res in result_vec {
+                                        match res {
+                                            Ok(ReturnSuccess::Value(Value {
+                                                value: UntaggedValue::Table(list),
+                                                ..
+                                            })) => {
+                                                for l in list {
+                                                    output.push(l);
+                                                }
+                                            }
+                                            Ok(ReturnSuccess::Value(Value { value, .. })) => {
+                                                output.push(value.into_value(contents_tag.clone()));
+                                            }
+                                            Err(e) => output.push(
+                                                UntaggedValue::Error(e).into_untagged_value(),
+                                            ),
+                                            _ => {}
+                                        }
+                                    }
+
+                                    let mut output = output.into_iter();
+
+                                    if let Some(x) = output.next() {
+                                        self.leftovers = output.collect();
+
+                                        return Some(x);
+                                    }
+                                }
+                                Err(err) => {
+                                    self.context.error(err);
+                                }
+                            }
+                        } else {
+                            return Some(tagged_contents);
+                        }
+                    }
+                    CommandAction::EnterValueShell(value) => {
+                        self.context
+                            .shell_manager
+                            .insert_at_current(Box::new(ValueShell::new(value)));
+                    }
+                    CommandAction::EnterShell(location) => {
+                        let mode = if self.context.shell_manager.is_interactive() {
+                            FilesystemShellMode::Cli
+                        } else {
+                            FilesystemShellMode::Script
+                        };
+                        self.context.shell_manager.insert_at_current(Box::new(
+                            match FilesystemShell::with_location(location, mode) {
+                                Ok(v) => v,
+                                Err(err) => {
+                                    self.context.error(err.into());
+                                    break;
+                                }
+                            },
+                        ));
+                    }
+                    CommandAction::AddPlugins(path) => {
+                        match crate::plugin::build_plugin::scan(vec![std::path::PathBuf::from(
+                            path,
+                        )]) {
+                            Ok(plugins) => {
+                                self.context.add_commands(
+                                    plugins
+                                        .into_iter()
+                                        .filter(|p| !self.context.is_command_registered(p.name()))
+                                        .collect(),
+                                );
+                            }
+                            Err(reason) => {
+                                self.context.error(reason);
+                            }
+                        }
+                    }
+                    CommandAction::PreviousShell => {
+                        self.context.shell_manager.prev();
+                    }
+                    CommandAction::NextShell => {
+                        self.context.shell_manager.next();
+                    }
+                    CommandAction::LeaveShell(code) => {
+                        self.context.shell_manager.remove_at_current();
+                        if self.context.shell_manager.is_empty() {
+                            std::process::exit(code); // TODO: save history.txt
+                        }
+                    }
+                    CommandAction::UnloadConfig(cfg_path) => {
+                        self.context.unload_config(&cfg_path);
+                    }
+                    CommandAction::LoadConfig(cfg_path) => {
+                        if let Err(e) = self.context.load_config(&cfg_path) {
+                            return Some(UntaggedValue::Error(e).into_untagged_value());
+                        }
+                    }
+                },
+
+                Ok(ReturnSuccess::Value(Value {
+                    value: UntaggedValue::Error(err),
+                    ..
+                })) => {
+                    self.context.error(err);
+                }
+
+                Ok(ReturnSuccess::Value(v)) => return Some(v),
+
+                Ok(ReturnSuccess::DebugValue(v)) => {
+                    let doc = PrettyDebug::pretty_doc(&v);
+                    let mut buffer = termcolor::Buffer::ansi();
+
+                    let _ = doc.render_raw(
+                        self.context.with_host(|host| host.width() - 5),
+                        &mut nu_source::TermColored::new(&mut buffer),
+                    );
+
+                    let value = String::from_utf8_lossy(buffer.as_slice());
+
+                    return Some(UntaggedValue::string(value).into_untagged_value());
+                }
+
+                Err(err) => {
+                    self.context.error(err);
+                }
+            }
+        }
+
+        None
+    }
+}

crates/nu-engine/src/filesystem/filesystem_shell.rs

@@ -778,13 +778,10 @@ impl Shell for FilesystemShell {
             }
         };
 
-        let mut stream = VecDeque::new();
+        Ok(OutputStream::one(ReturnSuccess::value(
-        stream.push_back(ReturnSuccess::value(
             UntaggedValue::Primitive(Primitive::String(p.to_string_lossy().to_string()))
                 .into_value(&args.call_info.name_tag),
-        ));
+        )))
-
-        Ok(stream.into())
     }
 
     fn set_path(&mut self, path: String) {

crates/nu-engine/src/shell/value_shell.rs

@@ -178,9 +178,7 @@ impl Shell for ValueShell {
             ));
         }
 
-        let mut stream = VecDeque::new();
+        Ok(OutputStream::one(ReturnSuccess::change_cwd(path)))
-        stream.push_back(ReturnSuccess::change_cwd(path));
-        Ok(stream.into())
     }
 
     fn cp(&self, _args: CopyArgs, name: Tag, _path: &str) -> Result<OutputStream, ShellError> {
@@ -220,11 +218,9 @@ impl Shell for ValueShell {
     }
 
     fn pwd(&self, args: EvaluatedWholeStreamCommandArgs) -> Result<OutputStream, ShellError> {
-        let mut stream = VecDeque::new();
+        Ok(OutputStream::one(
-        stream.push_back(ReturnSuccess::value(
             UntaggedValue::string(self.path()).into_value(&args.call_info.name_tag),
-        ));
+        ))
-        Ok(stream.into())
     }
 
     fn set_path(&mut self, path: String) {

crates/nu-stream/src/input.rs

@@ -27,9 +27,10 @@ impl InputStream {
     }
 
     pub fn one(item: impl Into<Value>) -> InputStream {
-        let mut v: VecDeque<Value> = VecDeque::new();
+        InputStream {
-        v.push_back(item.into());
+            values: Box::new(std::iter::once(item.into())),
-        v.into()
+            empty: false,
+        }
     }
 
     pub fn into_vec(self) -> Vec<Value> {

crates/nu-stream/src/prelude.rs

@@ -8,19 +8,6 @@ macro_rules! return_err {
     };
 }
 
-#[macro_export]
-macro_rules! stream {
-    ($($expr:expr),*) => {{
-        let mut v = VecDeque::new();
-
-        $(
-            v.push_back($expr);
-        )*
-
-        v
-    }}
-}
-
 #[macro_export]
 macro_rules! trace_out_stream {
     (target: $target:tt, $desc:tt = $expr:expr) => {{