rbt/scan/
mod.rs

1//! `rbt::scan`: Multi-format bronze lake scanners producing Arrow `RecordBatch`es.
2//!
3//! Formats: JSONL (jshift), JSON, Parquet, CSV, Arrow IPC (file/stream), log, txt, TOML.
4
5use crate::core::frontmatter::{resolve_scan_path, SourceFormat, StagingFrontmatter};
6use anyhow::{anyhow, bail, Context, Result};
7use arrow::array::{ArrayRef, Int64Builder, StringBuilder};
8use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
9use arrow::record_batch::RecordBatch;
10use std::fs::File;
11use std::io::{BufRead, BufReader};
12use std::path::{Path, PathBuf};
13use std::sync::Arc;
14
15/// Scan request derived from staging frontmatter + project root.
16#[derive(Debug, Clone)]
17pub struct ScanRequest {
18    pub project_dir: PathBuf,
19    pub scan_path: String,
20    pub format: SourceFormat,
21    /// jshift / projection paths (JSONL selective extract).
22    pub paths: Vec<String>,
23    /// TOML array-of-tables key (optional).
24    pub toml_rows_key: Option<String>,
25    /// Hive-style partition keys to inject as Utf8 columns from the file path.
26    pub partition_by: Vec<String>,
27    /// Keep only files whose hive path matches these partition values.
28    pub require_partitions: std::collections::HashMap<String, String>,
29    /// Inject `_source_path` column (absolute path of the bronze file).
30    pub inject_source_path: bool,
31}
32
33impl ScanRequest {
34    pub fn from_frontmatter(
35        project_dir: impl AsRef<Path>,
36        fm: &StagingFrontmatter,
37    ) -> Result<Self> {
38        let scan_path = fm
39            .scan_path
40            .as_ref()
41            .context("frontmatter.scan_path is required for bronze scan")?
42            .clone();
43        let format = fm.resolve_format()?;
44        Ok(Self {
45            project_dir: project_dir.as_ref().to_path_buf(),
46            scan_path,
47            format,
48            paths: fm.paths.clone().unwrap_or_default(),
49            toml_rows_key: fm.toml_rows_key.clone(),
50            partition_by: fm.partition_by.clone().unwrap_or_default(),
51            require_partitions: fm.require_partitions.clone().unwrap_or_default(),
52            inject_source_path: fm.inject_source_path.unwrap_or(false),
53        })
54    }
55
56    pub fn resolved_path(&self) -> PathBuf {
57        resolve_scan_path(&self.project_dir, &self.scan_path)
58    }
59}
60
61/// Multi-format lake scanner.
62pub struct LakeScanner {
63    pub paths: Vec<String>,
64}
65
66impl LakeScanner {
67    pub fn new(paths: Vec<String>) -> Self {
68        Self { paths }
69    }
70
71    pub fn from_request(req: &ScanRequest) -> Self {
72        Self {
73            paths: req.paths.clone(),
74        }
75    }
76
77    /// Scan using a full [`ScanRequest`] (format-aware).
78    pub async fn scan(&self, req: &ScanRequest) -> Result<Vec<RecordBatch>> {
79        let root = req.resolved_path();
80        if !root.exists() {
81            bail!(
82                "Bronze scan_path does not exist: {} (resolved from '{}')",
83                root.display(),
84                req.scan_path
85            );
86        }
87
88        let mut files = Vec::new();
89        collect_files_for_format(&root, req.format, &mut files)?;
90        if !req.require_partitions.is_empty() {
91            files.retain(|f| path_matches_require_partitions(f, &root, &req.require_partitions));
92        }
93        if files.is_empty() {
94            bail!(
95                "No {} files found under {} (after partition filters)",
96                req.format.as_str(),
97                root.display()
98            );
99        }
100
101        tracing::info!(
102            "Bronze scan: {} file(s) under {} (format={})",
103            files.len(),
104            root.display(),
105            req.format
106        );
107
108        let mut batches = Vec::new();
109        for file_path in files {
110            let file_batches = self
111                .read_file(&file_path, req)
112                .with_context(|| format!("Failed reading bronze file {}", file_path.display()))?;
113            for batch in file_batches {
114                let mut batch = if req.partition_by.is_empty() {
115                    batch
116                } else {
117                    inject_hive_partitions(batch, &file_path, &root, &req.partition_by)?
118                };
119                if req.inject_source_path {
120                    batch = inject_source_path_column(batch, &file_path)?;
121                }
122                batches.push(batch);
123            }
124        }
125        Ok(batches)
126    }
127
128    /// Legacy helper: recurse path and read by extension with an explicit schema
129    /// (used by unit tests and jshift-projected JSONL).
130    pub async fn scan_path(
131        &self,
132        path: impl AsRef<Path>,
133        schema: SchemaRef,
134    ) -> Result<Vec<RecordBatch>> {
135        let mut files = Vec::new();
136        collect_files(path.as_ref(), &mut files)?;
137        let mut batches = Vec::new();
138
139        for file_path in files {
140            let ext = file_path
141                .extension()
142                .and_then(|e| e.to_str())
143                .unwrap_or("")
144                .to_ascii_lowercase();
145            match ext.as_str() {
146                "parquet" => batches.extend(read_parquet(&file_path, Some(schema.clone()))?),
147                "json" | "jsonl" | "ndjson" => {
148                    let bytes = std::fs::read(&file_path)?;
149                    let extractor = crate::json::JShiftExtractor::new(self.paths.clone());
150                    batches.push(extractor.extract_jsonl(&bytes, schema.clone())?);
151                }
152                "csv" | "tsv" => batches.extend(read_csv(&file_path, Some(schema.clone()))?),
153                "log" | "txt" | "text" | "md" => {
154                    batches.push(read_line_oriented(&file_path)?);
155                }
156                "toml" => batches.push(read_toml(&file_path, None)?),
157                "arrow" | "arrows" | "ipc" | "feather" => {
158                    batches.extend(read_arrow_ipc_auto(&file_path)?);
159                }
160                _ => {
161                    tracing::debug!("Skipping unsupported file: {:?}", file_path);
162                }
163            }
164        }
165        Ok(batches)
166    }
167
168    fn read_file(&self, file_path: &Path, req: &ScanRequest) -> Result<Vec<RecordBatch>> {
169        match req.format {
170            SourceFormat::Parquet => read_parquet(file_path, None),
171            SourceFormat::Csv => read_csv(file_path, None),
172            SourceFormat::Jsonl | SourceFormat::Json => {
173                if !self.paths.is_empty() {
174                    let schema = utf8_schema_from_paths(&self.paths);
175                    let bytes = std::fs::read(file_path)?;
176                    let extractor = crate::json::JShiftExtractor::new(self.paths.clone());
177                    // JSONL extractor works line-wise; for single JSON array, expand first
178                    if req.format == SourceFormat::Json {
179                        let expanded = expand_json_document_to_jsonl(&bytes)?;
180                        Ok(vec![extractor.extract_jsonl(&expanded, schema)?])
181                    } else {
182                        Ok(vec![extractor.extract_jsonl(&bytes, schema)?])
183                    }
184                } else {
185                    // Schema-free path: use Arrow JSON reader (line-delimited)
186                    read_json_arrow(file_path, req.format)
187                }
188            }
189            // Real lakes often write stream IPC with a `.arrow` extension.
190            SourceFormat::ArrowIpc | SourceFormat::ArrowIpcStream => read_arrow_ipc_auto(file_path),
191            SourceFormat::Log | SourceFormat::Txt => Ok(vec![read_line_oriented(file_path)?]),
192            SourceFormat::Toml => Ok(vec![read_toml(file_path, req.toml_rows_key.as_deref())?]),
193        }
194    }
195}
196
197fn utf8_schema_from_paths(paths: &[String]) -> SchemaRef {
198    Arc::new(Schema::new(
199        paths
200            .iter()
201            .map(|p| Field::new(p.as_str(), DataType::Utf8, true))
202            .collect::<Vec<_>>(),
203    ))
204}
205
206fn expand_json_document_to_jsonl(bytes: &[u8]) -> Result<Vec<u8>> {
207    let v: serde_json::Value =
208        serde_json::from_slice(bytes).map_err(|e| anyhow!("Invalid JSON document: {}", e))?;
209    match v {
210        serde_json::Value::Array(items) => {
211            let mut out = Vec::new();
212            for item in items {
213                out.extend(serde_json::to_vec(&item)?);
214                out.push(b'\n');
215            }
216            Ok(out)
217        }
218        other => {
219            let mut out = serde_json::to_vec(&other)?;
220            out.push(b'\n');
221            Ok(out)
222        }
223    }
224}
225
226fn collect_files(path: &Path, files: &mut Vec<PathBuf>) -> Result<()> {
227    if path.is_file() {
228        files.push(path.to_path_buf());
229    } else if path.is_dir() {
230        for entry in std::fs::read_dir(path)? {
231            let entry = entry?;
232            collect_files(&entry.path(), files)?;
233        }
234    }
235    Ok(())
236}
237
238fn extension_matches(format: SourceFormat, ext: &str) -> bool {
239    let ext = ext.to_ascii_lowercase();
240    match format {
241        SourceFormat::Jsonl => matches!(ext.as_str(), "jsonl" | "ndjson"),
242        SourceFormat::Json => ext == "json",
243        SourceFormat::Parquet => matches!(ext.as_str(), "parquet" | "pq"),
244        SourceFormat::Csv => matches!(ext.as_str(), "csv" | "tsv"),
245        SourceFormat::ArrowIpc => matches!(ext.as_str(), "arrow" | "arrows" | "ipc" | "feather"),
246        SourceFormat::ArrowIpcStream => {
247            matches!(
248                ext.as_str(),
249                "arrow" | "arrows" | "ipc" | "arrows_stream" | "ipc_stream"
250            )
251        }
252        SourceFormat::Log => ext == "log",
253        SourceFormat::Txt => matches!(ext.as_str(), "txt" | "text" | "md"),
254        SourceFormat::Toml => ext == "toml",
255    }
256}
257
258fn collect_files_for_format(
259    path: &Path,
260    format: SourceFormat,
261    files: &mut Vec<PathBuf>,
262) -> Result<()> {
263    if path.is_file() {
264        files.push(path.to_path_buf());
265        return Ok(());
266    }
267    if path.is_dir() {
268        let mut all = Vec::new();
269        collect_files(path, &mut all)?;
270        for f in all {
271            let ext = f.extension().and_then(|e| e.to_str()).unwrap_or("");
272            if extension_matches(format, ext) {
273                files.push(f);
274            }
275        }
276        // If directory has files but none matched extension, and format is Log/Txt,
277        // still allow extensionless? Skip for now.
278        return Ok(());
279    }
280    bail!(
281        "scan path is neither file nor directory: {}",
282        path.display()
283    );
284}
285
286fn read_parquet(path: &Path, projection: Option<SchemaRef>) -> Result<Vec<RecordBatch>> {
287    let file = File::open(path)?;
288    let builder = parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(file)?;
289    let reader = if let Some(schema) = projection {
290        let schema_descr = builder.metadata().file_metadata().schema_descr_ptr();
291        let arrow_schema = builder.schema();
292        let mut indices = Vec::new();
293        for field in schema.fields() {
294            if let Some((idx, _)) =
295                parquet::arrow::parquet_column(&schema_descr, arrow_schema, field.name())
296            {
297                indices.push(idx);
298            }
299        }
300        let mask = parquet::arrow::ProjectionMask::leaves(&schema_descr, indices);
301        builder.with_projection(mask).build()?
302    } else {
303        builder.build()?
304    };
305    let mut batches = Vec::new();
306    for batch in reader {
307        batches.push(batch?);
308    }
309    Ok(batches)
310}
311
312fn read_csv(path: &Path, schema: Option<SchemaRef>) -> Result<Vec<RecordBatch>> {
313    let file = File::open(path)?;
314    let mut batches = Vec::new();
315    if let Some(schema) = schema {
316        let reader = arrow::csv::ReaderBuilder::new(schema)
317            .with_header(true)
318            .build(file)?;
319        for batch in reader {
320            batches.push(batch?);
321        }
322    } else {
323        // Infer schema from first rows
324        let format = arrow::csv::reader::Format::default().with_header(true);
325        let (inferred, _) = format.infer_schema(File::open(path)?, Some(1024))?;
326        let schema = Arc::new(inferred);
327        let file = File::open(path)?;
328        let reader = arrow::csv::ReaderBuilder::new(schema)
329            .with_header(true)
330            .build(file)?;
331        for batch in reader {
332            batches.push(batch?);
333        }
334    }
335    Ok(batches)
336}
337
338fn read_json_arrow(path: &Path, format: SourceFormat) -> Result<Vec<RecordBatch>> {
339    let data = std::fs::read(path)?;
340    let data = if format == SourceFormat::Json {
341        expand_json_document_to_jsonl(&data)?
342    } else {
343        data
344    };
345    read_json_infer_from_bytes(&data)
346}
347
348fn read_json_infer_from_bytes(data: &[u8]) -> Result<Vec<RecordBatch>> {
349    use arrow::json::reader::infer_json_schema_from_seekable;
350    let mut cursor = std::io::Cursor::new(data);
351    let (schema, _n) = infer_json_schema_from_seekable(&mut cursor, Some(1024))?;
352    cursor.set_position(0);
353    let schema = Arc::new(schema);
354    let reader = arrow::json::ReaderBuilder::new(schema).build(cursor)?;
355    let mut batches = Vec::new();
356    for batch in reader {
357        batches.push(batch?);
358    }
359    Ok(batches)
360}
361
362fn read_arrow_ipc_file(path: &Path) -> Result<Vec<RecordBatch>> {
363    let file = File::open(path)?;
364    let reader = arrow::ipc::reader::FileReader::try_new(file, None)?;
365    let mut batches = Vec::new();
366    for batch in reader {
367        batches.push(batch?);
368    }
369    Ok(batches)
370}
371
372fn read_arrow_ipc_stream(path: &Path) -> Result<Vec<RecordBatch>> {
373    let file = File::open(path)?;
374    let reader = arrow::ipc::reader::StreamReader::try_new(file, None)?;
375    let mut batches = Vec::new();
376    for batch in reader {
377        batches.push(batch?);
378    }
379    Ok(batches)
380}
381
382/// Prefer random-access IPC file; fall back to stream (common for `.arrow` lake dumps).
383fn read_arrow_ipc_auto(path: &Path) -> Result<Vec<RecordBatch>> {
384    match read_arrow_ipc_file(path) {
385        Ok(batches) => Ok(batches),
386        Err(file_err) => read_arrow_ipc_stream(path).with_context(|| {
387            format!(
388                "Arrow IPC file and stream readers both failed for {} (file error: {file_err})",
389                path.display()
390            )
391        }),
392    }
393}
394
395/// Parse `key=value` segments from `file` relative to `root`.
396pub fn parse_hive_partitions(
397    file: &Path,
398    root: &Path,
399) -> std::collections::HashMap<String, String> {
400    let mut out = std::collections::HashMap::new();
401    let rel = file.strip_prefix(root).unwrap_or(file);
402    for comp in rel.components() {
403        if let std::path::Component::Normal(os) = comp {
404            let s = os.to_string_lossy();
405            if let Some((k, v)) = s.split_once('=') {
406                if !k.is_empty() {
407                    out.insert(k.to_string(), v.to_string());
408                }
409            }
410        }
411    }
412    out
413}
414
415fn path_matches_require_partitions(
416    file: &Path,
417    root: &Path,
418    require: &std::collections::HashMap<String, String>,
419) -> bool {
420    if require.is_empty() {
421        return true;
422    }
423    let parts = parse_hive_partitions(file, root);
424    require
425        .iter()
426        .all(|(k, v)| parts.get(k).map(|pv| pv == v).unwrap_or(false))
427}
428
429/// Append hive partition columns (Utf8) for keys in `partition_by` that are not already present.
430fn inject_hive_partitions(
431    batch: RecordBatch,
432    file: &Path,
433    root: &Path,
434    partition_by: &[String],
435) -> Result<RecordBatch> {
436    if partition_by.is_empty() {
437        return Ok(batch);
438    }
439    let parts = parse_hive_partitions(file, root);
440    let n = batch.num_rows();
441    let mut fields: Vec<arrow::datatypes::Field> = batch
442        .schema()
443        .fields()
444        .iter()
445        .map(|f| f.as_ref().clone())
446        .collect();
447    let mut columns: Vec<ArrayRef> = batch.columns().to_vec();
448
449    for key in partition_by {
450        if batch.schema().index_of(key).is_ok() {
451            // Column already in payload (e.g. symbol); keep file data, do not overwrite.
452            continue;
453        }
454        let val = parts.get(key).map(|s| s.as_str());
455        let mut b = StringBuilder::with_capacity(n, n * 8);
456        for _ in 0..n {
457            match val {
458                Some(v) => b.append_value(v),
459                None => b.append_null(),
460            }
461        }
462        fields.push(Field::new(key.as_str(), DataType::Utf8, true));
463        columns.push(Arc::new(b.finish()) as ArrayRef);
464    }
465
466    let schema = Arc::new(Schema::new(fields));
467    Ok(RecordBatch::try_new(schema, columns)?)
468}
469
470/// Append `_source_path` Utf8 with the absolute bronze file path (for latest-wins dedupe).
471fn inject_source_path_column(batch: RecordBatch, file: &Path) -> Result<RecordBatch> {
472    if batch.schema().index_of("_source_path").is_ok() {
473        return Ok(batch);
474    }
475    let n = batch.num_rows();
476    let path_str = file.to_string_lossy();
477    let mut b = StringBuilder::with_capacity(n, n * path_str.len().max(16));
478    for _ in 0..n {
479        b.append_value(path_str.as_ref());
480    }
481    let mut fields: Vec<Field> = batch
482        .schema()
483        .fields()
484        .iter()
485        .map(|f| f.as_ref().clone())
486        .collect();
487    fields.push(Field::new("_source_path", DataType::Utf8, false));
488    let mut columns = batch.columns().to_vec();
489    columns.push(Arc::new(b.finish()) as ArrayRef);
490    Ok(RecordBatch::try_new(
491        Arc::new(Schema::new(fields)),
492        columns,
493    )?)
494}
495
496/// Line-oriented bronze: `.log`, `.txt`, llms.txt-style docs.
497/// Schema: `line_no` Int64, `content` Utf8.
498fn read_line_oriented(path: &Path) -> Result<RecordBatch> {
499    let file = File::open(path)?;
500    let reader = BufReader::new(file);
501    let mut line_nos = Int64Builder::new();
502    let mut contents = StringBuilder::new();
503    let mut n: i64 = 0;
504    for line in reader.lines() {
505        let line = line?;
506        n += 1;
507        line_nos.append_value(n);
508        contents.append_value(line);
509    }
510    let schema = Arc::new(Schema::new(vec![
511        Field::new("line_no", DataType::Int64, false),
512        Field::new("content", DataType::Utf8, false),
513    ]));
514    Ok(RecordBatch::try_new(
515        schema,
516        vec![
517            Arc::new(line_nos.finish()) as ArrayRef,
518            Arc::new(contents.finish()) as ArrayRef,
519        ],
520    )?)
521}
522
523fn read_toml(path: &Path, rows_key: Option<&str>) -> Result<RecordBatch> {
524    let text = std::fs::read_to_string(path)?;
525    let value: toml::Value = text
526        .parse()
527        .with_context(|| format!("Invalid TOML: {}", path.display()))?;
528
529    let rows: Vec<toml::map::Map<String, toml::Value>> = match &value {
530        toml::Value::Table(table) => {
531            if let Some(key) = rows_key {
532                match table.get(key) {
533                    Some(toml::Value::Array(arr)) => array_of_tables(arr)?,
534                    other => bail!(
535                        "toml_rows_key '{}' is not an array of tables (got {:?})",
536                        key,
537                        other.map(|v| v.type_str())
538                    ),
539                }
540            } else if let Some((_k, toml::Value::Array(arr))) = table
541                .iter()
542                .find(|(_, v)| matches!(v, toml::Value::Array(a) if a.iter().all(|x| x.is_table())))
543            {
544                array_of_tables(arr)?
545            } else {
546                // Single-row: top-level scalars / nested stringified
547                vec![table.clone()]
548            }
549        }
550        toml::Value::Array(arr) => array_of_tables(arr)?,
551        other => bail!("Unsupported top-level TOML type: {}", other.type_str()),
552    };
553
554    if rows.is_empty() {
555        let schema = Arc::new(Schema::new(vec![Field::new("empty", DataType::Utf8, true)]));
556        return Ok(RecordBatch::try_new(
557            schema,
558            vec![Arc::new(StringBuilder::new().finish()) as ArrayRef],
559        )?);
560    }
561
562    // Column union
563    let mut col_names: Vec<String> = Vec::new();
564    for row in &rows {
565        for k in row.keys() {
566            if !col_names.iter().any(|c| c == k) {
567                col_names.push(k.clone());
568            }
569        }
570    }
571
572    let mut builders: Vec<StringBuilder> = col_names
573        .iter()
574        .map(|_| StringBuilder::with_capacity(rows.len(), rows.len() * 16))
575        .collect();
576
577    for row in &rows {
578        for (i, col) in col_names.iter().enumerate() {
579            match row.get(col) {
580                Some(v) => builders[i].append_value(toml_value_to_string(v)),
581                None => builders[i].append_null(),
582            }
583        }
584    }
585
586    let fields: Vec<Field> = col_names
587        .iter()
588        .map(|c| Field::new(c.as_str(), DataType::Utf8, true))
589        .collect();
590    let schema = Arc::new(Schema::new(fields));
591    let arrays: Vec<ArrayRef> = builders
592        .into_iter()
593        .map(|mut b| Arc::new(b.finish()) as ArrayRef)
594        .collect();
595    Ok(RecordBatch::try_new(schema, arrays)?)
596}
597
598fn array_of_tables(arr: &[toml::Value]) -> Result<Vec<toml::map::Map<String, toml::Value>>> {
599    let mut rows = Vec::with_capacity(arr.len());
600    for (i, item) in arr.iter().enumerate() {
601        match item {
602            toml::Value::Table(t) => rows.push(t.clone()),
603            other => bail!(
604                "TOML array element {} is not a table (got {})",
605                i,
606                other.type_str()
607            ),
608        }
609    }
610    Ok(rows)
611}
612
613fn toml_value_to_string(v: &toml::Value) -> String {
614    match v {
615        toml::Value::String(s) => s.clone(),
616        toml::Value::Integer(i) => i.to_string(),
617        toml::Value::Float(f) => f.to_string(),
618        toml::Value::Boolean(b) => b.to_string(),
619        toml::Value::Datetime(d) => d.to_string(),
620        other => other.to_string(),
621    }
622}
623
624#[cfg(test)]
625mod tests {
626    use super::*;
627    use arrow::array::{Array, Int64Array, StringArray};
628    use arrow::datatypes::{DataType, Field, Schema};
629    use std::sync::Arc;
630
631    #[tokio::test]
632    async fn test_lake_scanner_multi_format() -> Result<()> {
633        let temp_dir = tempfile::tempdir()?;
634        let dir_path = temp_dir.path();
635
636        let schema = Arc::new(Schema::new(vec![
637            Field::new("id", DataType::Int64, true),
638            Field::new("name", DataType::Utf8, true),
639        ]));
640
641        std::fs::write(
642            dir_path.join("file1.jsonl"),
643            b"{\"id\": 1, \"name\": \"Alice\"}\n{\"id\": 2, \"name\": \"Bob\"}\n",
644        )?;
645        std::fs::write(dir_path.join("file2.csv"), b"id,name\n3,Charlie\n4,Dave\n")?;
646
647        let batch = RecordBatch::try_new(
648            schema.clone(),
649            vec![
650                Arc::new(Int64Array::from(vec![5, 6])),
651                Arc::new(StringArray::from(vec!["Eve", "Frank"])),
652            ],
653        )?;
654        let file = File::create(dir_path.join("file3.parquet"))?;
655        let mut writer =
656            parquet::arrow::arrow_writer::ArrowWriter::try_new(file, schema.clone(), None)?;
657        writer.write(&batch)?;
658        writer.close()?;
659
660        std::fs::write(dir_path.join("app.log"), "info boot\nwarn disk\n")?;
661        std::fs::write(
662            dir_path.join("meta.toml"),
663            r#"
664[[records]]
665id = "1"
666name = "toml_a"
667[[records]]
668id = "2"
669name = "toml_b"
670"#,
671        )?;
672
673        let scanner = LakeScanner::new(vec!["id".to_string(), "name".to_string()]);
674        let batches = scanner.scan_path(dir_path, schema.clone()).await?;
675        let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
676        // 2 jsonl + 2 csv + 2 parquet + 2 log + 2 toml = 10
677        assert_eq!(total_rows, 10);
678        Ok(())
679    }
680
681    #[tokio::test]
682    async fn test_scan_request_jsonl() -> Result<()> {
683        let temp = tempfile::tempdir()?;
684        let path = temp.path().join("trades.jsonl");
685        std::fs::write(
686            &path,
687            r#"{"ticker":"NVDA","price":1.0}
688{"ticker":"AAPL","price":2.0}
689"#,
690        )?;
691
692        let fm = StagingFrontmatter {
693            source_format: Some(SourceFormat::Jsonl),
694            scan_path: Some(path.file_name().unwrap().to_string_lossy().into()),
695            ..Default::default()
696        };
697        let req = ScanRequest::from_frontmatter(temp.path(), &fm)?;
698        let scanner = LakeScanner::from_request(&req);
699        let batches = scanner.scan(&req).await?;
700        let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
701        assert_eq!(rows, 2);
702        Ok(())
703    }
704
705    #[tokio::test]
706    async fn test_hive_partition_inject_and_filter() -> Result<()> {
707        let temp = tempfile::tempdir()?;
708        let root = temp.path();
709        let dir = root.join("symbol=NVDA").join("timeframe=1m");
710        std::fs::create_dir_all(&dir)?;
711        // minimal IPC stream with one utf8 column
712        let schema = Arc::new(Schema::new(vec![Field::new(
713            "close",
714            DataType::Float64,
715            true,
716        )]));
717        let batch = RecordBatch::try_new(
718            schema,
719            vec![Arc::new(arrow::array::Float64Array::from(vec![1.0, 2.0])) as ArrayRef],
720        )?;
721        let path = dir.join("chunk.arrow");
722        {
723            let file = File::create(&path)?;
724            let mut writer = arrow::ipc::writer::StreamWriter::try_new(file, &batch.schema())?;
725            writer.write(&batch)?;
726            writer.finish()?;
727        }
728        // noise partition that should be filtered out
729        let other = root.join("symbol=AAPL").join("timeframe=1d");
730        std::fs::create_dir_all(&other)?;
731        {
732            let file = File::create(other.join("chunk.arrow"))?;
733            let mut writer = arrow::ipc::writer::StreamWriter::try_new(file, &batch.schema())?;
734            writer.write(&batch)?;
735            writer.finish()?;
736        }
737
738        let mut require = std::collections::HashMap::new();
739        require.insert("timeframe".into(), "1m".into());
740        let fm = StagingFrontmatter {
741            source_format: Some(SourceFormat::ArrowIpcStream),
742            scan_path: Some(".".into()),
743            partition_by: Some(vec!["symbol".into(), "timeframe".into()]),
744            require_partitions: Some(require),
745            ..Default::default()
746        };
747        let req = ScanRequest::from_frontmatter(root, &fm)?;
748        let scanner = LakeScanner::from_request(&req);
749        let batches = scanner.scan(&req).await?;
750        let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
751        assert_eq!(rows, 2);
752        let schema = batches[0].schema();
753        assert!(schema.index_of("timeframe").is_ok());
754        assert!(schema.index_of("symbol").is_ok());
755        Ok(())
756    }
757
758    #[test]
759    fn test_line_oriented_and_toml() -> Result<()> {
760        let temp = tempfile::tempdir()?;
761        let txt = temp.path().join("llms.txt");
762        std::fs::write(&txt, "# Title\n\nSome doc line\n")?;
763        let batch = read_line_oriented(&txt)?;
764        assert_eq!(batch.num_rows(), 3);
765        assert_eq!(batch.schema().field(0).name(), "line_no");
766        assert_eq!(batch.schema().field(1).name(), "content");
767
768        let toml_path = temp.path().join("cfg.toml");
769        std::fs::write(
770            &toml_path,
771            r#"
772[[items]]
773k = "a"
774[[items]]
775k = "b"
776"#,
777        )?;
778        let tbatch = read_toml(&toml_path, Some("items"))?;
779        assert_eq!(tbatch.num_rows(), 2);
780        let col = tbatch
781            .column(0)
782            .as_any()
783            .downcast_ref::<StringArray>()
784            .unwrap();
785        assert!(col.value(0) == "a" || col.value(1) == "a");
786        Ok(())
787    }
788}
rbt/scan/mod.rs

rbt/scan/
mod.rs
