laminar_connectors/files/text_decoder/
mod.rs

1//! Plain text line decoder implementing [`FormatDecoder`].
2//!
3//! Splits raw bytes by newlines and produces a single-column `RecordBatch`
4//! with column `line: Utf8`.
5
6use std::sync::Arc;
7
8use arrow_array::{RecordBatch, StringArray};
9use arrow_schema::{DataType, Field, Schema, SchemaRef};
10
11use crate::schema::error::{SchemaError, SchemaResult};
12use crate::schema::traits::FormatDecoder;
13use crate::schema::types::RawRecord;
14
15/// Decodes raw bytes as newline-delimited text into a single `line` column.
16#[derive(Debug)]
17pub struct TextLineDecoder {
18    schema: SchemaRef,
19}
20
21impl TextLineDecoder {
22    /// Creates a new text line decoder.
23    #[must_use]
24    pub fn new() -> Self {
25        let schema = Arc::new(Schema::new(vec![Field::new("line", DataType::Utf8, false)]));
26        Self { schema }
27    }
28}
29
30impl Default for TextLineDecoder {
31    fn default() -> Self {
32        Self::new()
33    }
34}
35
36impl FormatDecoder for TextLineDecoder {
37    fn output_schema(&self) -> SchemaRef {
38        self.schema.clone()
39    }
40
41    fn decode_batch(&self, records: &[RawRecord]) -> SchemaResult<RecordBatch> {
42        if records.is_empty() {
43            return Ok(RecordBatch::new_empty(self.schema.clone()));
44        }
45
46        // Validate UTF-8 upfront, then collect &str slices to avoid per-line allocation.
47        let mut texts = Vec::with_capacity(records.len());
48        for record in records {
49            let text = std::str::from_utf8(&record.value).map_err(|e| {
50                SchemaError::DecodeError(format!("invalid UTF-8 in text file: {e}"))
51            })?;
52            texts.push(text);
53        }
54
55        let lines: Vec<&str> = texts
56            .iter()
57            .flat_map(|t| t.lines())
58            .filter(|l| !l.is_empty())
59            .collect();
60        let array = StringArray::from_iter_values(lines);
61        RecordBatch::try_new(self.schema.clone(), vec![Arc::new(array)])
62            .map_err(|e| SchemaError::DecodeError(format!("batch construction error: {e}")))
63    }
64
65    fn format_name(&self) -> &'static str {
66        "text"
67    }
68}
69#[cfg(test)]
70mod tests;
laminar_connectors/files/text_decoder/mod.rs

laminar_connectors/files/text_decoder/
mod.rs
