diff --git a/crates/nu-command/src/conversions/into/binary.rs b/crates/nu-command/src/conversions/into/binary.rs
index 667bcb9eee..6c635f49da 100644
--- a/crates/nu-command/src/conversions/into/binary.rs
+++ b/crates/nu-command/src/conversions/into/binary.rs
@@ -103,7 +103,7 @@ fn into_binary(
             // TODO: in the future, we may want this to stream out, converting each to bytes
             let output = stream.into_bytes()?;
             Ok(Value::Binary {
-                val: output,
+                val: output.item,
                 span: head,
             }
             .into_pipeline_data())
diff --git a/crates/nu-command/src/conversions/into/string.rs b/crates/nu-command/src/conversions/into/string.rs
index 7d40f6da88..e57b114726 100644
--- a/crates/nu-command/src/conversions/into/string.rs
+++ b/crates/nu-command/src/conversions/into/string.rs
@@ -154,7 +154,7 @@ fn string_helper(
             // TODO: in the future, we may want this to stream out, converting each to bytes
             let output = stream.into_string()?;
             Ok(Value::String {
-                val: output,
+                val: output.item,
                 span: head,
             }
             .into_pipeline_data())
diff --git a/crates/nu-command/src/strings/decode.rs b/crates/nu-command/src/strings/decode.rs
index cd2de9354d..7b54791b5f 100644
--- a/crates/nu-command/src/strings/decode.rs
+++ b/crates/nu-command/src/strings/decode.rs
@@ -45,7 +45,7 @@ impl Command for Decode {
 
         match input {
             PipelineData::RawStream(stream, ..) => {
-                let bytes: Vec<u8> = stream.into_bytes()?;
+                let bytes: Vec<u8> = stream.into_bytes()?.item;
 
                 let encoding = match Encoding::for_label(encoding.item.as_bytes()) {
                     None => Err(ShellError::SpannedLabeledError(
diff --git a/crates/nu-protocol/src/pipeline_data.rs b/crates/nu-protocol/src/pipeline_data.rs
index c095d25bef..c048bf5b21 100644
--- a/crates/nu-protocol/src/pipeline_data.rs
+++ b/crates/nu-protocol/src/pipeline_data.rs
@@ -230,12 +230,23 @@ impl PipelineData {
                 Ok(vals.into_iter().map(f).into_pipeline_data(ctrlc))
             }
             PipelineData::ListStream(stream, ..) => Ok(stream.map(f).into_pipeline_data(ctrlc)),
-            PipelineData::RawStream(stream, ..) => Ok(stream
-                .map(move |x| match x {
-                    Ok(v) => f(v),
-                    Err(err) => Value::Error { error: err },
-                })
-                .into_pipeline_data(ctrlc)),
+            PipelineData::RawStream(stream, ..) => {
+                let collected = stream.into_bytes()?;
+
+                if let Ok(st) = String::from_utf8(collected.clone().item) {
+                    Ok(f(Value::String {
+                        val: st,
+                        span: collected.span,
+                    })
+                    .into_pipeline_data())
+                } else {
+                    Ok(f(Value::Binary {
+                        val: collected.item,
+                        span: collected.span,
+                    })
+                    .into_pipeline_data())
+                }
+            }
 
             PipelineData::Value(Value::Range { val, .. }, ..) => {
                 Ok(val.into_range_iter()?.map(f).into_pipeline_data(ctrlc))
@@ -266,14 +277,25 @@ impl PipelineData {
             PipelineData::ListStream(stream, ..) => {
                 Ok(stream.map(f).flatten().into_pipeline_data(ctrlc))
             }
-            PipelineData::RawStream(stream, ..) => Ok(stream
-                .map(move |x| match x {
-                    Ok(v) => v,
-                    Err(err) => Value::Error { error: err },
-                })
-                .map(f)
-                .flatten()
-                .into_pipeline_data(ctrlc)),
+            PipelineData::RawStream(stream, ..) => {
+                let collected = stream.into_bytes()?;
+
+                if let Ok(st) = String::from_utf8(collected.clone().item) {
+                    Ok(f(Value::String {
+                        val: st,
+                        span: collected.span,
+                    })
+                    .into_iter()
+                    .into_pipeline_data(ctrlc))
+                } else {
+                    Ok(f(Value::Binary {
+                        val: collected.item,
+                        span: collected.span,
+                    })
+                    .into_iter()
+                    .into_pipeline_data(ctrlc))
+                }
+            }
             PipelineData::Value(Value::Range { val, .. }, ..) => match val.into_range_iter() {
                 Ok(iter) => Ok(iter.map(f).flatten().into_pipeline_data(ctrlc)),
                 Err(error) => Err(error),
@@ -296,13 +318,33 @@ impl PipelineData {
                 Ok(vals.into_iter().filter(f).into_pipeline_data(ctrlc))
             }
             PipelineData::ListStream(stream, ..) => Ok(stream.filter(f).into_pipeline_data(ctrlc)),
-            PipelineData::RawStream(stream, ..) => Ok(stream
-                .map(move |x| match x {
-                    Ok(v) => v,
-                    Err(err) => Value::Error { error: err },
-                })
-                .filter(f)
-                .into_pipeline_data(ctrlc)),
+            PipelineData::RawStream(stream, ..) => {
+                let collected = stream.into_bytes()?;
+
+                if let Ok(st) = String::from_utf8(collected.clone().item) {
+                    let v = Value::String {
+                        val: st,
+                        span: collected.span,
+                    };
+
+                    if f(&v) {
+                        Ok(v.into_pipeline_data())
+                    } else {
+                        Ok(PipelineData::new(collected.span))
+                    }
+                } else {
+                    let v = Value::Binary {
+                        val: collected.item,
+                        span: collected.span,
+                    };
+
+                    if f(&v) {
+                        Ok(v.into_pipeline_data())
+                    } else {
+                        Ok(PipelineData::new(collected.span))
+                    }
+                }
+            }
             PipelineData::Value(Value::Range { val, .. }, ..) => {
                 Ok(val.into_range_iter()?.filter(f).into_pipeline_data(ctrlc))
             }
diff --git a/crates/nu-protocol/src/value/stream.rs b/crates/nu-protocol/src/value/stream.rs
index 7478b7c221..6f00299eb7 100644
--- a/crates/nu-protocol/src/value/stream.rs
+++ b/crates/nu-protocol/src/value/stream.rs
@@ -30,24 +30,28 @@ impl RawStream {
         }
     }
 
-    pub fn into_bytes(self) -> Result<Vec<u8>, ShellError> {
+    pub fn into_bytes(self) -> Result<Spanned<Vec<u8>>, ShellError> {
         let mut output = vec![];
 
         for item in self.stream {
             output.extend(item?);
         }
 
-        Ok(output)
+        Ok(Spanned {
+            item: output,
+            span: self.span,
+        })
     }
 
-    pub fn into_string(self) -> Result<String, ShellError> {
+    pub fn into_string(self) -> Result<Spanned<String>, ShellError> {
         let mut output = String::new();
+        let span = self.span;
 
         for item in self {
             output.push_str(&item?.as_string()?);
         }
 
-        Ok(output)
+        Ok(Spanned { item: output, span })
     }
 }
 impl Debug for RawStream {