oxidelake_runtime/
output.rs1use std::io::Write;
10use std::str::FromStr;
11
12use datafusion::arrow::array::RecordBatch;
13use datafusion::arrow::datatypes::SchemaRef;
14use datafusion::arrow::util::pretty::pretty_format_batches;
15use oxidelake_core::EngineError;
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
19#[non_exhaustive]
20pub enum OutputFormat {
21 #[default]
23 Table,
24 Json,
26 Csv,
28}
29
30impl OutputFormat {
31 pub const ALL: [OutputFormat; 3] = [OutputFormat::Table, OutputFormat::Json, OutputFormat::Csv];
33
34 pub const fn as_str(self) -> &'static str {
36 match self {
37 OutputFormat::Table => "table",
38 OutputFormat::Json => "json",
39 OutputFormat::Csv => "csv",
40 }
41 }
42}
43
44impl std::fmt::Display for OutputFormat {
45 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46 f.write_str(self.as_str())
47 }
48}
49
50impl FromStr for OutputFormat {
51 type Err = EngineError;
52
53 fn from_str(s: &str) -> Result<Self, Self::Err> {
54 match s.trim().to_ascii_lowercase().as_str() {
55 "table" => Ok(OutputFormat::Table),
56 "json" => Ok(OutputFormat::Json),
57 "csv" => Ok(OutputFormat::Csv),
58 other => Err(EngineError::plan(format!(
59 "unknown output format '{other}'; expected one of table, json, csv"
60 ))),
61 }
62 }
63}
64
65pub fn render(
71 schema: &SchemaRef,
72 batches: &[RecordBatch],
73 format: OutputFormat,
74) -> Result<String, EngineError> {
75 match format {
76 OutputFormat::Table => Ok(format!("{}\n", pretty_format_batches(batches)?)),
77 OutputFormat::Json => {
78 let mut buffer = Vec::new();
79 let mut writer = datafusion::arrow::json::ArrayWriter::new(&mut buffer);
80 writer.write_batches(&batches.iter().collect::<Vec<_>>())?;
81 writer.finish()?;
82 finish(buffer)
83 }
84 OutputFormat::Csv => {
85 let mut buffer = Vec::new();
86 let mut writer = datafusion::arrow::csv::Writer::new(&mut buffer);
87 let empty;
88 let batches = if batches.is_empty() {
89 empty = [RecordBatch::new_empty(SchemaRef::clone(schema))];
90 &empty[..]
91 } else {
92 batches
93 };
94 for batch in batches {
95 writer.write(batch)?;
96 }
97 drop(writer);
98 finish(buffer)
99 }
100 }
101}
102
103fn finish(mut buffer: Vec<u8>) -> Result<String, EngineError> {
105 if !buffer.ends_with(b"\n") {
106 buffer.write_all(b"\n").map_err(EngineError::from)?;
107 }
108 String::from_utf8(buffer)
109 .map_err(|e| EngineError::execution(format!("rendered output was not UTF-8: {e}")))
110}
111
112#[cfg(test)]
113#[allow(clippy::unwrap_used, clippy::expect_used)]
114mod tests {
115 use std::sync::Arc;
116
117 use datafusion::arrow::array::{Int64Array, StringArray};
118 use datafusion::arrow::datatypes::{DataType, Field, Schema};
119
120 use super::*;
121
122 fn schema() -> SchemaRef {
123 Arc::new(Schema::new(vec![
124 Field::new("k", DataType::Int64, true),
125 Field::new("name", DataType::Utf8, true),
126 ]))
127 }
128
129 fn batch() -> RecordBatch {
130 RecordBatch::try_new(
131 schema(),
132 vec![
133 Arc::new(Int64Array::from(vec![Some(1), None])),
134 Arc::new(StringArray::from(vec![Some("a,b"), None])),
135 ],
136 )
137 .unwrap()
138 }
139
140 #[test]
141 fn every_name_round_trips() {
142 for format in OutputFormat::ALL {
143 assert_eq!(format.as_str().parse::<OutputFormat>().unwrap(), format);
144 }
145 assert_eq!(" CSV ".parse::<OutputFormat>().unwrap(), OutputFormat::Csv);
146 }
147
148 #[test]
149 fn an_unknown_name_names_the_alternatives() {
150 let message = "yaml".parse::<OutputFormat>().unwrap_err().to_string();
151 assert!(message.contains("yaml"), "{message}");
152 assert!(message.contains("table, json, csv"), "{message}");
153 }
154
155 #[test]
156 fn json_is_an_array_of_objects_and_drops_nulls() {
157 let text = render(&schema(), &[batch()], OutputFormat::Json).unwrap();
158 assert_eq!(text, "[{\"k\":1,\"name\":\"a,b\"},{}]\n");
159 }
160
161 #[test]
162 fn csv_quotes_an_embedded_comma_and_keeps_the_header() {
163 let text = render(&schema(), &[batch()], OutputFormat::Csv).unwrap();
164 assert_eq!(text, "k,name\n1,\"a,b\"\n,\n");
165 }
166
167 #[test]
170 fn an_empty_result_still_has_a_csv_header() {
171 assert_eq!(
172 render(&schema(), &[], OutputFormat::Csv).unwrap(),
173 "k,name\n"
174 );
175 assert_eq!(render(&schema(), &[], OutputFormat::Json).unwrap(), "[]\n");
176 }
177
178 #[test]
179 fn the_table_form_is_the_default() {
180 assert_eq!(OutputFormat::default(), OutputFormat::Table);
181 let text = render(&schema(), &[batch()], OutputFormat::Table).unwrap();
182 assert!(text.contains("| k "), "{text}");
183 assert!(text.ends_with("\n"), "{text}");
184 }
185}