# Description This PR introduces a `ByteStream` type which is a `Read`-able stream of bytes. Internally, it has an enum over three different byte stream sources: ```rust pub enum ByteStreamSource { Read(Box<dyn Read + Send + 'static>), File(File), Child(ChildProcess), } ``` This is in comparison to the current `RawStream` type, which is an `Iterator<Item = Vec<u8>>` and has to allocate for each read chunk. Currently, `PipelineData::ExternalStream` serves a weird dual role where it is either external command output or a wrapper around `RawStream`. `ByteStream` makes this distinction more clear (via `ByteStreamSource`) and replaces `PipelineData::ExternalStream` in this PR: ```rust pub enum PipelineData { Empty, Value(Value, Option<PipelineMetadata>), ListStream(ListStream, Option<PipelineMetadata>), ByteStream(ByteStream, Option<PipelineMetadata>), } ``` The PR is relatively large, but a decent amount of it is just repetitive changes. This PR fixes #7017, fixes #10763, and fixes #12369. This PR also improves performance when piping external commands. Nushell should, in most cases, have competitive pipeline throughput compared to, e.g., bash. | Command | Before (MB/s) | After (MB/s) | Bash (MB/s) | | -------------------------------------------------- | -------------:| ------------:| -----------:| | `throughput \| rg 'x'` | 3059 | 3744 | 3739 | | `throughput \| nu --testbin relay o> /dev/null` | 3508 | 8087 | 8136 | # User-Facing Changes - This is a breaking change for the plugin communication protocol, because the `ExternalStreamInfo` was replaced with `ByteStreamInfo`. Plugins now only have to deal with a single input stream, as opposed to the previous three streams: stdout, stderr, and exit code. - The output of `describe` has been changed for external/byte streams. - Temporary breaking change: `bytes starts-with` no longer works with byte streams. This is to keep the PR smaller, and `bytes ends-with` already does not work on byte streams. - If a process core dumped, then instead of having a `Value::Error` in the `exit_code` column of the output returned from `complete`, it now is a `Value::Int` with the negation of the signal number. # After Submitting - Update docs and book as necessary - Release notes (e.g., plugin protocol changes) - Adapt/convert commands to work with byte streams (high priority is `str length`, `bytes starts-with`, and maybe `bytes ends-with`). - Refactor the `tee` code, Devyn has already done some work on this. --------- Co-authored-by: Devyn Cairns <devyn.cairns@gmail.com>
123 lines
3.8 KiB
Rust
123 lines
3.8 KiB
Rust
use crate::dataframe::values::{Column, NuDataFrame, NuExpression};
|
|
use nu_engine::command_prelude::*;
|
|
use polars::prelude::IntoSeries;
|
|
|
|
#[derive(Clone)]
|
|
pub struct IsNotNull;
|
|
|
|
impl Command for IsNotNull {
|
|
fn name(&self) -> &str {
|
|
"dfr is-not-null"
|
|
}
|
|
|
|
fn usage(&self) -> &str {
|
|
"Creates mask where value is not null."
|
|
}
|
|
|
|
fn signature(&self) -> Signature {
|
|
Signature::build(self.name())
|
|
.input_output_types(vec![
|
|
(
|
|
Type::Custom("expression".into()),
|
|
Type::Custom("expression".into()),
|
|
),
|
|
(
|
|
Type::Custom("dataframe".into()),
|
|
Type::Custom("dataframe".into()),
|
|
),
|
|
])
|
|
.category(Category::Custom("dataframe".into()))
|
|
}
|
|
|
|
fn examples(&self) -> Vec<Example> {
|
|
vec![
|
|
Example {
|
|
description: "Create mask where values are not null",
|
|
example: r#"let s = ([5 6 0 8] | dfr into-df);
|
|
let res = ($s / $s);
|
|
$res | dfr is-not-null"#,
|
|
result: Some(
|
|
NuDataFrame::try_from_columns(
|
|
vec![Column::new(
|
|
"is_not_null".to_string(),
|
|
vec![
|
|
Value::test_bool(true),
|
|
Value::test_bool(true),
|
|
Value::test_bool(false),
|
|
Value::test_bool(true),
|
|
],
|
|
)],
|
|
None,
|
|
)
|
|
.expect("simple df for test should not fail")
|
|
.into_value(Span::test_data()),
|
|
),
|
|
},
|
|
Example {
|
|
description: "Creates a is not null expression from a column",
|
|
example: "dfr col a | dfr is-not-null",
|
|
result: None,
|
|
},
|
|
]
|
|
}
|
|
|
|
fn run(
|
|
&self,
|
|
engine_state: &EngineState,
|
|
stack: &mut Stack,
|
|
call: &Call,
|
|
input: PipelineData,
|
|
) -> Result<PipelineData, ShellError> {
|
|
let value = input.into_value(call.head)?;
|
|
if NuDataFrame::can_downcast(&value) {
|
|
let df = NuDataFrame::try_from_value(value)?;
|
|
command(engine_state, stack, call, df)
|
|
} else {
|
|
let expr = NuExpression::try_from_value(value)?;
|
|
let expr: NuExpression = expr.into_polars().is_not_null().into();
|
|
|
|
Ok(PipelineData::Value(
|
|
NuExpression::into_value(expr, call.head),
|
|
None,
|
|
))
|
|
}
|
|
}
|
|
}
|
|
|
|
fn command(
|
|
_engine_state: &EngineState,
|
|
_stack: &mut Stack,
|
|
call: &Call,
|
|
df: NuDataFrame,
|
|
) -> Result<PipelineData, ShellError> {
|
|
let mut res = df.as_series(call.head)?.is_not_null();
|
|
res.rename("is_not_null");
|
|
|
|
NuDataFrame::try_from_series(vec![res.into_series()], call.head)
|
|
.map(|df| PipelineData::Value(NuDataFrame::into_value(df, call.head), None))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod test {
|
|
use super::*;
|
|
use crate::dataframe::lazy::aggregate::LazyAggregate;
|
|
use crate::dataframe::lazy::groupby::ToLazyGroupBy;
|
|
use crate::dataframe::test_dataframe::{build_test_engine_state, test_dataframe_example};
|
|
|
|
#[test]
|
|
fn test_examples_dataframe() {
|
|
let mut engine_state = build_test_engine_state(vec![Box::new(IsNotNull {})]);
|
|
test_dataframe_example(&mut engine_state, &IsNotNull.examples()[0]);
|
|
}
|
|
|
|
#[test]
|
|
fn test_examples_expression() {
|
|
let mut engine_state = build_test_engine_state(vec![
|
|
Box::new(IsNotNull {}),
|
|
Box::new(LazyAggregate {}),
|
|
Box::new(ToLazyGroupBy {}),
|
|
]);
|
|
test_dataframe_example(&mut engine_state, &IsNotNull.examples()[1]);
|
|
}
|
|
}
|