Merge pull request #592 from jonathandturner/table_to

Covert to_* commands to work on whole table

src/cli.rs

@@ -179,7 +179,6 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
             whole_stream_command(Reject),
             whole_stream_command(Reverse),
             whole_stream_command(Trim),
-            whole_stream_command(ToArray),
             whole_stream_command(ToBSON),
             whole_stream_command(ToCSV),
             whole_stream_command(ToJSON),
@@ -192,8 +191,6 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
             whole_stream_command(Tags),
             whole_stream_command(First),
             whole_stream_command(Last),
-            whole_stream_command(FromArray),
-            whole_stream_command(FromArray),
             whole_stream_command(FromCSV),
             whole_stream_command(FromTSV),
             whole_stream_command(FromINI),

src/commands.rs

@@ -15,7 +15,6 @@ pub(crate) mod enter;
 pub(crate) mod exit;
 pub(crate) mod fetch;
 pub(crate) mod first;
-pub(crate) mod from_array;
 pub(crate) mod from_bson;
 pub(crate) mod from_csv;
 pub(crate) mod from_ini;
@@ -52,7 +51,6 @@ pub(crate) mod split_column;
 pub(crate) mod split_row;
 pub(crate) mod table;
 pub(crate) mod tags;
-pub(crate) mod to_array;
 pub(crate) mod to_bson;
 pub(crate) mod to_csv;
 pub(crate) mod to_json;
@@ -81,7 +79,6 @@ pub(crate) use enter::Enter;
 pub(crate) use exit::Exit;
 pub(crate) use fetch::Fetch;
 pub(crate) use first::First;
-pub(crate) use from_array::FromArray;
 pub(crate) use from_bson::FromBSON;
 pub(crate) use from_csv::FromCSV;
 pub(crate) use from_ini::FromINI;
@@ -119,7 +116,6 @@ pub(crate) use split_column::SplitColumn;
 pub(crate) use split_row::SplitRow;
 pub(crate) use table::Table;
 pub(crate) use tags::Tags;
-pub(crate) use to_array::ToArray;
 pub(crate) use to_bson::ToBSON;
 pub(crate) use to_csv::ToCSV;
 pub(crate) use to_json::ToJSON;

src/commands/from_array.rs

@@ -1,43 +0,0 @@
-use crate::commands::WholeStreamCommand;
-use crate::object::Value;
-use crate::prelude::*;
-
-pub struct FromArray;
-
-impl WholeStreamCommand for FromArray {
-    fn name(&self) -> &str {
-        "from-array"
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::build("from-array")
-    }
-
-    fn usage(&self) -> &str {
-        "Expand an array/list into rows"
-    }
-
-    fn run(
-        &self,
-        args: CommandArgs,
-        registry: &CommandRegistry,
-    ) -> Result<OutputStream, ShellError> {
-        from_array(args, registry)
-    }
-}
-
-fn from_array(args: CommandArgs, _registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let stream = args
-        .input
-        .values
-        .map(|item| match item {
-            Tagged {
-                item: Value::List(vec),
-                ..
-            } => VecDeque::from(vec),
-            x => VecDeque::from(vec![x]),
-        })
-        .flatten();
-
-    Ok(stream.to_output_stream())
-}

src/commands/to_array.rs

@@ -1,38 +0,0 @@
-use crate::commands::WholeStreamCommand;
-use crate::object::Value;
-use crate::prelude::*;
-
-pub struct ToArray;
-
-impl WholeStreamCommand for ToArray {
-    fn name(&self) -> &str {
-        "to-array"
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::build("to-array")
-    }
-
-    fn usage(&self) -> &str {
-        "Collapse rows into a single list."
-    }
-
-    fn run(
-        &self,
-        args: CommandArgs,
-        registry: &CommandRegistry,
-    ) -> Result<OutputStream, ShellError> {
-        to_array(args, registry)
-    }
-}
-
-fn to_array(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
-    let args = args.evaluate_once(registry)?;
-    let span = args.call_info.name_span;
-    let out = args.input.values.collect();
-
-    Ok(out
-        .map(move |vec: Vec<_>| stream![Value::List(vec).simple_spanned(span)])
-        .flatten_stream()
-        .from_input_stream())
-}

src/commands/to_bson.rs

@@ -236,21 +236,41 @@ fn bson_value_to_bytes(bson: Bson, span: Span) -> Result<Vec<u8>, ShellError> {
 fn to_bson(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
-    let out = args.input;
+    let stream = async_stream_block! {
+        let input: Vec<Tagged<Value>> = args.input.values.collect().await;
 
-    Ok(out
+        let to_process_input = if input.len() > 1 {
-        .values
+            let tag = input[0].tag;
-        .map(
+            vec![Tagged { item: Value::List(input), tag } ]
-            move |a| match bson_value_to_bytes(value_to_bson_value(&a)?, name_span) {
+        } else if input.len() == 1 {
-                Ok(x) => ReturnSuccess::value(Value::Binary(x).simple_spanned(name_span)),
+            input
-                _ => Err(ShellError::labeled_error_with_secondary(
+        } else {
-                    "Expected an object with BSON-compatible structure from pipeline",
+            vec![]
-                    "requires BSON-compatible input: Must be Array or Object",
+        };
+
+        for value in to_process_input {
+            match value_to_bson_value(&value) {
+                Ok(bson_value) => {
+                    match bson_value_to_bytes(bson_value, name_span) {
+                        Ok(x) => yield ReturnSuccess::value(
+                            Value::Binary(x).simple_spanned(name_span),
+                        ),
+                        _ => yield Err(ShellError::labeled_error_with_secondary(
+                            "Expected an object with BSON-compatible structure.span() from pipeline",
+                            "requires BSON-compatible input",
                             name_span,
-                    format!("{} originates from here", a.item.type_name()),
+                            "originates from here".to_string(),
-                    a.span(),
+                            value.span(),
                         )),
-            },
+                    }
-        )
+                }
-        .to_output_stream())
+                _ => yield Err(ShellError::labeled_error(
+                    "Expected an object with BSON-compatible structure from pipeline",
+                    "requires BSON-compatible input",
+                    name_span))
+            }
+        }
+    };
+
+    Ok(stream.to_output_stream())
 }

src/commands/to_csv.rs

@@ -16,8 +16,7 @@ impl WholeStreamCommand for ToCSV {
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("to-csv")
+        Signature::build("to-csv").switch("headerless")
-            .switch("headerless")
     }
 
     fn usage(&self) -> &str {
@@ -47,7 +46,7 @@ pub fn value_to_csv_value(v: &Value) -> Value {
     }
 }
 
-fn to_string_helper(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
+fn to_string_helper(v: &Value) -> Result<String, ShellError> {
     match v {
         Value::Primitive(Primitive::Date(d)) => Ok(d.to_string()),
         Value::Primitive(Primitive::Bytes(b)) => Ok(format!("{}", b)),
@@ -55,11 +54,23 @@ fn to_string_helper(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
         Value::List(_) => return Ok(String::from("[list list]")),
         Value::Object(_) => return Ok(String::from("[object]")),
         Value::Primitive(Primitive::String(s)) => return Ok(s.to_string()),
-        _ => return Err("Bad input".into()),
+        _ => return Err(ShellError::string("Unexpected value")),
     }
 }
 
-pub fn to_string(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
+fn merge_descriptors(values: &[Tagged<Value>]) -> Vec<String> {
+    let mut ret = vec![];
+    for value in values {
+        for desc in value.data_descriptors() {
+            if !ret.contains(&desc) {
+                ret.push(desc);
+            }
+        }
+    }
+    ret
+}
+
+pub fn to_string(v: &Value) -> Result<String, ShellError> {
     match v {
         Value::Object(o) => {
             let mut wtr = WriterBuilder::new().from_writer(vec![]);
@@ -68,13 +79,46 @@ pub fn to_string(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
 
             for (k, v) in o.entries.iter() {
                 fields.push_back(k.clone());
+
                 values.push_back(to_string_helper(&v)?);
             }
 
             wtr.write_record(fields).expect("can not write.");
             wtr.write_record(values).expect("can not write.");
 
-            return Ok(String::from_utf8(wtr.into_inner()?)?);
+            return Ok(String::from_utf8(
+                wtr.into_inner()
+                    .map_err(|_| ShellError::string("Could not convert record"))?,
+            )
+            .map_err(|_| ShellError::string("Could not convert record"))?);
+        }
+        Value::List(list) => {
+            let mut wtr = WriterBuilder::new().from_writer(vec![]);
+
+            let merged_descriptors = merge_descriptors(&list);
+            wtr.write_record(&merged_descriptors)
+                .expect("can not write.");
+
+            for l in list {
+                let mut row = vec![];
+                for desc in &merged_descriptors {
+                    match l.item.get_data_by_key(&desc) {
+                        Some(s) => {
+                            row.push(to_string_helper(s)?);
+                        }
+                        None => {
+                            row.push(String::new());
+                        }
+                    }
+                }
+                wtr.write_record(&row).expect("can not write");
+            }
+
+            return Ok(String::from_utf8(
+                wtr.into_inner()
+                    .map_err(|_| ShellError::string("Could not convert record"))?,
+            )
+            .map_err(|_| ShellError::string("Could not convert record"))?);
         }
         _ => return to_string_helper(&v),
     }
@@ -85,29 +129,40 @@ fn to_csv(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let name_span = name;
-    let out = input;
+    let stream = async_stream_block! {
+         let input: Vec<Tagged<Value>> = input.values.collect().await;
 
-    Ok(out
+         let to_process_input = if input.len() > 1 {
-        .values
+             let tag = input[0].tag;
-        .map(move |a| match to_string(&value_to_csv_value(&a.item)) {
+             vec![Tagged { item: Value::List(input), tag } ]
+         } else if input.len() == 1 {
+             input
+         } else {
+             vec![]
+         };
+
+         for value in to_process_input {
+             match to_string(&value_to_csv_value(&value.item)) {
                  Ok(x) => {
                      let converted = if headerless {
                          x.lines().skip(1).collect()
                      } else {
                          x
                      };
-
+                     yield ReturnSuccess::value(Value::Primitive(Primitive::String(converted)).simple_spanned(name_span))
-                ReturnSuccess::value(
-                    Value::Primitive(Primitive::String(converted)).simple_spanned(name_span),
-                )
                  }
-            _ => Err(ShellError::labeled_error_with_secondary(
+                 _ => {
-                "Expected an object with CSV-compatible structure from pipeline",
+                     yield Err(ShellError::labeled_error_with_secondary(
+                         "Expected an object with CSV-compatible structure.span() from pipeline",
                          "requires CSV-compatible input",
                          name_span,
-                format!("{} originates from here", a.item.type_name()),
+                         "originates from here".to_string(),
-                a.span(),
+                         value.span(),
-            )),
+                     ))
-        })
+                 }
-        .to_output_stream())
+             }
+         }
+    };
+
+    Ok(stream.to_output_stream())
 }

src/commands/to_json.rs

@@ -80,23 +80,41 @@ fn json_list(input: &Vec<Tagged<Value>>) -> Result<Vec<serde_json::Value>, Shell
 fn to_json(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
-    let out = args.input;
+    let stream = async_stream_block! {
+        let input: Vec<Tagged<Value>> = args.input.values.collect().await;
 
-    Ok(out
+        let to_process_input = if input.len() > 1 {
-        .values
+            let tag = input[0].tag;
-        .map(
+            vec![Tagged { item: Value::List(input), tag } ]
-            move |a| match serde_json::to_string(&value_to_json_value(&a)?) {
+        } else if input.len() == 1 {
-                Ok(x) => ReturnSuccess::value(
+            input
+        } else {
+            vec![]
+        };
+
+        for value in to_process_input {
+            match value_to_json_value(&value) {
+                Ok(json_value) => {
+                    match serde_json::to_string(&json_value) {
+                        Ok(x) => yield ReturnSuccess::value(
                             Value::Primitive(Primitive::String(x)).simple_spanned(name_span),
                         ),
-                _ => Err(ShellError::labeled_error_with_secondary(
+                        _ => yield Err(ShellError::labeled_error_with_secondary(
-                    "Expected an object with JSON-compatible structure from pipeline",
+                            "Expected an object with JSON-compatible structure.span() from pipeline",
                             "requires JSON-compatible input",
                             name_span,
-                    format!("{} originates from here", a.item.type_name()),
+                            "originates from here".to_string(),
-                    a.span(),
+                            value.span(),
                         )),
-            },
+                    }
-        )
+                }
-        .to_output_stream())
+                _ => yield Err(ShellError::labeled_error(
+                    "Expected an object with JSON-compatible structure from pipeline",
+                    "requires JSON-compatible input",
+                    name_span))
+            }
+        }
+    };
+
+    Ok(stream.to_output_stream())
 }

src/commands/to_sqlite.rs

@@ -201,19 +201,19 @@ fn to_sqlite(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStre
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
     let stream = async_stream_block! {
-        let values: Vec<_> = args.input.into_vec().await;
+        let input: Vec<Tagged<Value>> = args.input.values.collect().await;
-        match sqlite_input_stream_to_bytes(values) {
+
-            Ok(out) => {
+        match sqlite_input_stream_to_bytes(input) {
-                yield ReturnSuccess::value(out)
+            Ok(out) => yield ReturnSuccess::value(out),
-            }
+            _ => {
-            Err(_) => {
                 yield Err(ShellError::labeled_error(
-                    "Expected an object with SQLite-compatible structure from pipeline",
+                    "Expected an object with SQLite-compatible structure.span() from pipeline",
                     "requires SQLite-compatible input",
                     name_span,
                 ))
+            },
         }
     };
-    };
+
     Ok(stream.to_output_stream())
 }

src/commands/to_toml.rs

@@ -75,23 +75,41 @@ fn collect_values(input: &Vec<Tagged<Value>>) -> Result<Vec<toml::Value>, ShellE
 fn to_toml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
-    let out = args.input;
+    let stream = async_stream_block! {
+        let input: Vec<Tagged<Value>> = args.input.values.collect().await;
 
-    Ok(out
+        let to_process_input = if input.len() > 1 {
-        .values
+            let tag = input[0].tag;
-        .map(move |a| match toml::to_string(&value_to_toml_value(&a)?) {
+            vec![Tagged { item: Value::List(input), tag } ]
-            Ok(val) => {
+        } else if input.len() == 1 {
-                return ReturnSuccess::value(
+            input
-                    Value::Primitive(Primitive::String(val)).simple_spanned(name_span),
+        } else {
-                )
+            vec![]
-            }
+        };
-            _ => Err(ShellError::labeled_error_with_secondary(
+
-                "Expected an object with TOML-compatible structure from pipeline",
+        for value in to_process_input {
+            match value_to_toml_value(&value) {
+                Ok(toml_value) => {
+                    match toml::to_string(&toml_value) {
+                        Ok(x) => yield ReturnSuccess::value(
+                            Value::Primitive(Primitive::String(x)).simple_spanned(name_span),
+                        ),
+                        _ => yield Err(ShellError::labeled_error_with_secondary(
+                            "Expected an object with TOML-compatible structure.span() from pipeline",
                             "requires TOML-compatible input",
                             name_span,
-                format!("{} originates from here", a.item.type_name()),
+                            "originates from here".to_string(),
-                a.span(),
+                            value.span(),
                         )),
-        })
+                    }
-        .to_output_stream())
+                }
+                _ => yield Err(ShellError::labeled_error(
+                    "Expected an object with TOML-compatible structure from pipeline",
+                    "requires TOML-compatible input",
+                    name_span))
+            }
+        }
+    };
+
+    Ok(stream.to_output_stream())
 }

src/commands/to_tsv.rs

@@ -16,8 +16,7 @@ impl WholeStreamCommand for ToTSV {
     }
 
     fn signature(&self) -> Signature {
-        Signature::build("to-tsv")
+        Signature::build("to-tsv").switch("headerless")
-            .switch("headerless")
     }
 
     fn usage(&self) -> &str {
@@ -47,7 +46,7 @@ pub fn value_to_tsv_value(v: &Value) -> Value {
     }
 }
 
-fn to_string_helper(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
+fn to_string_helper(v: &Value) -> Result<String, ShellError> {
     match v {
         Value::Primitive(Primitive::Date(d)) => Ok(d.to_string()),
         Value::Primitive(Primitive::Bytes(b)) => Ok(format!("{}", b)),
@@ -55,11 +54,23 @@ fn to_string_helper(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
         Value::List(_) => return Ok(String::from("[list list]")),
         Value::Object(_) => return Ok(String::from("[object]")),
         Value::Primitive(Primitive::String(s)) => return Ok(s.to_string()),
-        _ => return Err("Bad input".into()),
+        _ => Err(ShellError::string("Unexpected value")),
     }
 }
 
-pub fn to_string(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
+fn merge_descriptors(values: &[Tagged<Value>]) -> Vec<String> {
+    let mut ret = vec![];
+    for value in values {
+        for desc in value.data_descriptors() {
+            if !ret.contains(&desc) {
+                ret.push(desc);
+            }
+        }
+    }
+    ret
+}
+
+pub fn to_string(v: &Value) -> Result<String, ShellError> {
     match v {
         Value::Object(o) => {
             let mut wtr = WriterBuilder::new().delimiter(b'\t').from_writer(vec![]);
@@ -74,7 +85,39 @@ pub fn to_string(v: &Value) -> Result<String, Box<dyn std::error::Error>> {
             wtr.write_record(fields).expect("can not write.");
             wtr.write_record(values).expect("can not write.");
 
-            return Ok(String::from_utf8(wtr.into_inner()?)?);
+            return Ok(String::from_utf8(
+                wtr.into_inner()
+                    .map_err(|_| ShellError::string("Could not convert record"))?,
+            )
+            .map_err(|_| ShellError::string("Could not convert record"))?);
+        }
+        Value::List(list) => {
+            let mut wtr = WriterBuilder::new().delimiter(b'\t').from_writer(vec![]);
+
+            let merged_descriptors = merge_descriptors(&list);
+            wtr.write_record(&merged_descriptors)
+                .expect("can not write.");
+
+            for l in list {
+                let mut row = vec![];
+                for desc in &merged_descriptors {
+                    match l.item.get_data_by_key(&desc) {
+                        Some(s) => {
+                            row.push(to_string_helper(s)?);
+                        }
+                        None => {
+                            row.push(String::new());
+                        }
+                    }
+                }
+                wtr.write_record(&row).expect("can not write");
+            }
+
+            return Ok(String::from_utf8(
+                wtr.into_inner()
+                    .map_err(|_| ShellError::string("Could not convert record"))?,
+            )
+            .map_err(|_| ShellError::string("Could not convert record"))?);
         }
         _ => return to_string_helper(&v),
     }
@@ -85,29 +128,40 @@ fn to_tsv(
     RunnableContext { input, name, .. }: RunnableContext,
 ) -> Result<OutputStream, ShellError> {
     let name_span = name;
-    let out = input;
+    let stream = async_stream_block! {
+         let input: Vec<Tagged<Value>> = input.values.collect().await;
 
-    Ok(out
+         let to_process_input = if input.len() > 1 {
-        .values
+             let tag = input[0].tag;
-        .map(move |a| match to_string(&value_to_tsv_value(&a.item)) {
+             vec![Tagged { item: Value::List(input), tag } ]
+         } else if input.len() == 1 {
+             input
+         } else {
+             vec![]
+         };
+
+         for value in to_process_input {
+             match to_string(&value_to_tsv_value(&value.item)) {
                  Ok(x) => {
                      let converted = if headerless {
                          x.lines().skip(1).collect()
                      } else {
                          x
                      };
-
+                     yield ReturnSuccess::value(Value::Primitive(Primitive::String(converted)).simple_spanned(name_span))
-                ReturnSuccess::value(
-                    Value::Primitive(Primitive::String(converted)).simple_spanned(name_span),
-                )
                  }
-            _ => Err(ShellError::labeled_error_with_secondary(
+                 _ => {
-                "Expected an object with TSV-compatible structure from pipeline",
+                     yield Err(ShellError::labeled_error_with_secondary(
+                         "Expected an object with TSV-compatible structure.span() from pipeline",
                          "requires TSV-compatible input",
                          name_span,
-                format!("{} originates from here", a.item.type_name()),
+                         "originates from here".to_string(),
-                a.span(),
+                         value.span(),
-            )),
+                     ))
-        })
+                 }
-        .to_output_stream())
+             }
+         }
+    };
+
+    Ok(stream.to_output_stream())
 }

src/commands/to_yaml.rs

@@ -76,22 +76,41 @@ pub fn value_to_yaml_value(v: &Tagged<Value>) -> Result<serde_yaml::Value, Shell
 fn to_yaml(args: CommandArgs, registry: &CommandRegistry) -> Result<OutputStream, ShellError> {
     let args = args.evaluate_once(registry)?;
     let name_span = args.name_span();
-    let out = args.input;
+    let stream = async_stream_block! {
-    Ok(out
+        let input: Vec<Tagged<Value>> = args.input.values.collect().await;
-        .values
+
-        .map(
+        let to_process_input = if input.len() > 1 {
-            move |a| match serde_yaml::to_string(&value_to_yaml_value(&a)?) {
+            let tag = input[0].tag;
-                Ok(x) => ReturnSuccess::value(
+            vec![Tagged { item: Value::List(input), tag } ]
+        } else if input.len() == 1 {
+            input
+        } else {
+            vec![]
+        };
+
+        for value in to_process_input {
+            match value_to_yaml_value(&value) {
+                Ok(yaml_value) => {
+                    match serde_yaml::to_string(&yaml_value) {
+                        Ok(x) => yield ReturnSuccess::value(
                             Value::Primitive(Primitive::String(x)).simple_spanned(name_span),
                         ),
-                _ => Err(ShellError::labeled_error_with_secondary(
+                        _ => yield Err(ShellError::labeled_error_with_secondary(
-                    "Expected an object with YAML-compatible structure from pipeline",
+                            "Expected an object with YAML-compatible structure.span() from pipeline",
                             "requires YAML-compatible input",
                             name_span,
-                    format!("{} originates from here", a.item.type_name()),
+                            "originates from here".to_string(),
-                    a.span(),
+                            value.span(),
                         )),
-            },
+                    }
-        )
+                }
-        .to_output_stream())
+                _ => yield Err(ShellError::labeled_error(
+                    "Expected an object with YAML-compatible structure from pipeline",
+                    "requires YAML-compatible input",
+                    name_span))
+            }
+        }
+    };
+
+    Ok(stream.to_output_stream())
 }

tests/filters_test.rs

@@ -218,8 +218,8 @@ fn converts_structured_table_to_json_text() {
                 | split-column "," name luck
                 | pick name
                 | to-json
-                | nth 0
                 | from-json
+                | nth 0
                 | get name
                 | echo $it
             "#