1use rudb_common::{Error, Field, LogicalType, Result, Value};
13use rudb_io::File;
14use rudb_kernels::cast_value;
15use rudb_vector::{Chunk, VECTOR_SIZE, Vector};
16
17use crate::dialect::{self, Dialect, Given};
18use crate::infer;
19
20const BLOCK: usize = 1 << 20;
26
27#[derive(Debug)]
29pub struct Reader {
30 file: Box<dyn File>,
31 path: String,
32 given: Given,
33 dialect: Dialect,
34 fields: Vec<Field>,
35 projection: Vec<usize>,
36 buffer: Vec<u8>,
37 at: usize,
38 offset: u64,
39 drained: bool,
40 line: u64,
41 scratch: Vec<String>,
42}
43
44impl Reader {
45 pub fn open(file: Box<dyn File>, path: &str) -> Result<Self> {
54 Self::open_with(file, path, Given::default())
55 }
56
57 pub fn open_with(file: Box<dyn File>, path: &str, given: Given) -> Result<Self> {
67 let mut reader = Self {
68 file,
69 path: path.to_string(),
70 given,
71 dialect: Dialect::comma_separated(),
72 fields: Vec::new(),
73 projection: Vec::new(),
74 buffer: Vec::new(),
75 at: 0,
76 offset: 0,
77 drained: false,
78 line: 1,
79 scratch: Vec::new(),
80 };
81 reader.fill()?;
82 let sample = reader.buffer.clone();
83 let quote = given.quote.or_else(|| dialect::quote(&sample));
84 let delimiter = match given.delimiter {
85 Some(byte) => byte,
86 None => dialect::delimiter(&sample, quote)?,
87 };
88 let escape = given.escape.or(quote);
89 reader.dialect = Dialect { delimiter, quote, escape, header: false };
90 let rows = reader.sample_rows(&sample)?;
91 let (header, fields) = describe(&rows, given.header);
92 reader.dialect.header = header;
93 reader.fields = fields;
94 reader.projection = (0..reader.fields.len()).collect();
95 if header {
96 reader.skip_record()?;
97 }
98 Ok(reader)
99 }
100
101 #[must_use]
103 pub fn fields(&self) -> Vec<Field> {
104 self.projection.iter().map(|&at| self.fields[at].clone()).collect()
105 }
106
107 pub fn project(&mut self, columns: &[usize]) -> Result<()> {
113 for &column in columns {
114 if column >= self.fields.len() {
115 return Err(Error::io(format!(
116 "column {column} is past the {} the file has",
117 self.fields.len()
118 )));
119 }
120 }
121 self.projection = columns.to_vec();
122 Ok(())
123 }
124
125 pub fn retype(&mut self, types: &[LogicalType]) -> Result<()> {
143 if types.len() != self.projection.len() {
144 return Err(Error::io(format!(
145 "{} types for a projection of {} columns",
146 types.len(),
147 self.projection.len()
148 )));
149 }
150 for (&at, ty) in self.projection.iter().zip(types) {
151 self.fields[at].ty = ty.clone();
152 }
153 Ok(())
154 }
155
156 #[must_use]
158 pub const fn dialect(&self) -> Dialect {
159 self.dialect
160 }
161
162 pub fn next_chunk(&mut self) -> Result<Option<Chunk>> {
169 let mut rows: Vec<Vec<Option<String>>> = Vec::new();
170 while rows.len() < VECTOR_SIZE {
171 match self.next_record()? {
172 Some(fields) => rows.push(fields),
173 None => break,
174 }
175 }
176 if rows.is_empty() {
177 return Ok(None);
178 }
179 let mut columns = Vec::with_capacity(self.projection.len());
180 for &at in &self.projection {
181 let field = &self.fields[at];
182 let mut values = Vec::with_capacity(rows.len());
183 for (row, held) in rows.iter().enumerate() {
184 let text = held.get(at).and_then(Option::as_deref);
185 values.push(self.convert(
186 text,
187 field,
188 self.line - rows.len() as u64 + row as u64,
189 )?);
190 }
191 columns.push(Vector::from_values(field.ty.clone(), &values)?);
192 }
193 Ok(Some(Chunk::with_rows(columns, rows.len())?))
194 }
195
196 fn convert(&self, text: Option<&str>, field: &Field, line: u64) -> Result<Value> {
198 let Some(text) = text else { return Ok(Value::Null) };
199 if field.ty == LogicalType::Varchar {
200 return Ok(Value::Varchar(text.to_string()));
201 }
202 let value = Value::Varchar(text.to_string());
203 match cast_value(&value, &field.ty, false) {
204 Ok(converted) => Ok(converted),
205 Err(_) => Err(Error::conversion(self.conversion_error(text, field, line))),
206 }
207 }
208
209 fn conversion_error(&self, text: &str, field: &Field, line: u64) -> String {
221 format!(
222 "CSV Error on Line: {line}\nOriginal Line: {text}\nError when converting column \
223 \"{}\". Could not convert string \"{text}\" to '{}'\n\nColumn {} is being converted \
224 as type {}\nThis type was auto-detected from the CSV file.\nPossible solutions:\n* \
225 Override the type for this column manually by setting the type explicitly, e.g., \
226 types={{'{}': 'VARCHAR'}}\n* Set the sample size to a larger value to enable the \
227 auto-detection to scan more values, e.g., sample_size=-1\n* Use a COPY statement to \
228 automatically derive types from an existing table.\n* Check whether the null string \
229 value is set correctly (e.g., nullstr = 'N/A')\n\n file = {}\n delimiter = {}\n \
230 quote = {}\n escape = {}\n header = {} {}\n sample_size = {}\n",
231 field.name,
232 field.ty,
233 field.name,
234 field.ty,
235 field.name,
236 self.path,
237 Given::shown(self.given.delimiter, Some(self.dialect.delimiter)),
238 Given::shown(self.given.quote, self.dialect.quote),
239 Given::shown(self.given.escape, self.dialect.escape),
240 self.dialect.header,
241 Given::source(self.given.header.is_some()),
242 infer::SAMPLE,
243 )
244 }
245
246 fn next_record(&mut self) -> Result<Option<Vec<Option<String>>>> {
248 let Some(()) = self.advance()? else { return Ok(None) };
249 Ok(Some(
250 self.scratch
251 .iter()
252 .map(|text| if text.is_empty() { None } else { Some(text.clone()) })
253 .collect(),
254 ))
255 }
256
257 fn advance(&mut self) -> Result<Option<()>> {
259 loop {
260 let mut scratch = std::mem::take(&mut self.scratch);
261 let outcome = crate::scan::record(
262 &self.buffer,
263 self.at,
264 self.dialect,
265 self.drained,
266 &mut scratch,
267 );
268 self.scratch = scratch;
269 match outcome? {
270 Some(next) => {
271 self.at = next;
272 self.line += 1;
273 return Ok(Some(()));
274 }
275 None if self.drained => return Ok(None),
276 None => self.fill()?,
277 }
278 }
279 }
280
281 fn skip_record(&mut self) -> Result<()> {
283 self.advance()?;
284 Ok(())
285 }
286
287 fn fill(&mut self) -> Result<()> {
289 self.buffer.drain(..self.at);
290 self.at = 0;
291 let held = self.buffer.len();
292 self.buffer.resize(held + BLOCK, 0);
293 let read = self.file.read_at(self.offset, &mut self.buffer[held..])?;
294 self.buffer.truncate(held + read);
295 self.offset += read as u64;
296 if read == 0 {
297 self.drained = true;
298 }
299 Ok(())
300 }
301
302 fn sample_rows(&self, sample: &[u8]) -> Result<Vec<Vec<Option<String>>>> {
305 let mut rows = Vec::new();
306 let mut fields = Vec::new();
307 let mut at = 0;
308 while rows.len() <= infer::SAMPLE {
309 let Some(next) = crate::scan::record(sample, at, self.dialect, false, &mut fields)?
312 else {
313 break;
314 };
315 at = next;
316 rows.push(
317 fields
318 .iter()
319 .map(|text| if text.is_empty() { None } else { Some(text.clone()) })
320 .collect(),
321 );
322 }
323 Ok(rows)
324 }
325}
326
327fn describe(rows: &[Vec<Option<String>>], told: Option<bool>) -> (bool, Vec<Field>) {
338 let width = rows.iter().map(Vec::len).max().unwrap_or(0);
339 let body = types(&rows[1.min(rows.len())..], width);
340 let all_text = body.iter().all(|ty| *ty == LogicalType::Varchar);
341 let first_fits = rows.first().is_some_and(|first| {
342 first.iter().zip(&body).all(|(text, ty)| match text {
343 None => true,
344 Some(text) => infer::fits(text, ty),
345 })
346 });
347 let header = !rows.is_empty() && told.unwrap_or(rows.len() > 1 && (all_text || !first_fits));
349 if !header {
350 let types = types(rows, width);
351 let fields = types
352 .into_iter()
353 .enumerate()
354 .map(|(at, ty)| Field::new(format!("column{at}"), ty))
355 .collect();
356 return (false, fields);
357 }
358 let names = unique(&rows[0], width);
359 let fields = body.into_iter().zip(names).map(|(ty, name)| Field::new(name, ty)).collect();
360 (true, fields)
361}
362
363fn unique(header: &[Option<String>], width: usize) -> Vec<String> {
376 let mut taken: Vec<String> = Vec::with_capacity(width);
377 for at in 0..width {
378 let base = match header.get(at).and_then(Option::as_deref) {
379 Some(written) => written.to_string(),
380 None => format!("column{at}"),
381 };
382 let mut name = base.clone();
383 let mut next = 1;
384 while taken.iter().any(|held| held.eq_ignore_ascii_case(&name)) {
385 name = format!("{base}_{next}");
386 next += 1;
387 }
388 taken.push(name);
389 }
390 taken
391}
392
393fn types(rows: &[Vec<Option<String>>], width: usize) -> Vec<LogicalType> {
395 (0..width)
396 .map(|at| {
397 let values: Vec<Option<&str>> =
398 rows.iter().map(|row| row.get(at).and_then(Option::as_deref)).collect();
399 infer::column(&values)
400 })
401 .collect()
402}
403
404#[cfg(test)]
405mod tests {
406 use super::*;
407 use rudb_io::{Filesystem, OpenMode, SimFilesystem};
408 use std::path::Path;
409
410 fn read(text: &str) -> Reader {
411 let filesystem = SimFilesystem::new();
412 let path = Path::new("/t.csv");
413 let file = filesystem.open(path, OpenMode::Create).expect("creates");
414 file.write_at(0, text.as_bytes()).expect("writes");
415 drop(file);
416 let file = filesystem.open(path, OpenMode::Read).expect("opens");
417 Reader::open(file, "/t.csv").expect("sniffs")
418 }
419
420 fn read_with(text: &str, given: Given) -> Reader {
422 let filesystem = SimFilesystem::new();
423 let path = Path::new("/t.csv");
424 let file = filesystem.open(path, OpenMode::Create).expect("creates");
425 file.write_at(0, text.as_bytes()).expect("writes");
426 drop(file);
427 let file = filesystem.open(path, OpenMode::Read).expect("opens");
428 Reader::open_with(file, "/t.csv", given).expect("reads")
429 }
430
431 fn names_and_types(reader: &Reader) -> Vec<(String, String)> {
432 reader.fields().iter().map(|f| (f.name.clone(), f.ty.to_string())).collect()
433 }
434
435 fn all(reader: &mut Reader) -> Vec<Vec<Value>> {
436 let mut rows = Vec::new();
437 while let Some(chunk) = reader.next_chunk().expect("reads") {
438 for row in 0..chunk.len() {
439 rows.push((0..chunk.width()).map(|at| chunk.value_at(row, at)).collect());
440 }
441 }
442 rows
443 }
444
445 #[test]
446 fn a_header_that_names_two_columns_the_same_thing_counts_the_second_one_up() {
447 let names: Vec<String> =
448 read("a,a,A,a_1\n1,2,3,4\nx,y,z,w\n").fields().into_iter().map(|f| f.name).collect();
449 assert_eq!(names, ["a", "a_1", "A_2", "a_1_1"]);
453 }
454
455 #[test]
456 fn a_header_and_three_types_are_what_duckdb_sniffs_for_the_same_bytes() {
457 let reader = read("a,b,c\n1,x,2.5\n2,y,3.5\n");
458 assert_eq!(
459 names_and_types(&reader),
460 [
461 ("a".to_string(), "BIGINT".to_string()),
462 ("b".to_string(), "VARCHAR".to_string()),
463 ("c".to_string(), "DOUBLE".to_string()),
464 ]
465 );
466 }
467
468 #[test]
469 fn a_file_with_no_header_gets_the_names_duckdb_gives_it() {
470 let reader = read("1,x\n2,y\n");
471 assert_eq!(
472 names_and_types(&reader),
473 [
474 ("column0".to_string(), "BIGINT".to_string()),
475 ("column1".to_string(), "VARCHAR".to_string()),
476 ]
477 );
478 }
479
480 #[test]
481 fn two_rows_of_words_are_a_header_and_a_row() {
482 let reader = read("a,b\nc,d\n");
483 assert_eq!(reader.fields().iter().map(|f| f.name.clone()).collect::<Vec<_>>(), ["a", "b"]);
484 }
485
486 #[test]
487 fn one_column_of_words_under_a_row_of_numbers_is_still_a_header() {
488 let reader = read("a,2\n3,4\n");
491 assert_eq!(reader.fields().iter().map(|f| f.name.clone()).collect::<Vec<_>>(), ["a", "2"]);
492 }
493
494 #[test]
495 fn the_rows_are_the_rows_of_the_file() {
496 let mut reader = read("a,b\n1,x\n2,y\n");
497 assert_eq!(
498 all(&mut reader),
499 [
500 vec![Value::BigInt(1), Value::Varchar("x".into())],
501 vec![Value::BigInt(2), Value::Varchar("y".into())],
502 ]
503 );
504 }
505
506 #[test]
507 fn an_empty_field_is_a_null_whether_it_was_quoted_or_not() {
508 let mut reader = read("a,b\n1,\n\"\",y\n");
511 assert_eq!(
512 all(&mut reader),
513 [vec![Value::BigInt(1), Value::Null], vec![Value::Null, Value::Varchar("y".into())],]
514 );
515 }
516
517 #[test]
518 fn a_projection_picks_columns_out_by_position_and_can_reorder_them() {
519 let mut reader = read("a,b,c\n1,x,2.5\n");
520 reader.project(&[2, 0]).expect("projects");
521 assert_eq!(reader.fields().iter().map(|f| f.name.clone()).collect::<Vec<_>>(), ["c", "a"]);
522 assert_eq!(all(&mut reader), [vec![Value::Double(2.5), Value::BigInt(1)]]);
523 }
524
525 #[test]
526 fn a_projection_of_nothing_still_counts_the_rows() {
527 let mut reader = read("a,b\n1,x\n2,y\n3,z\n");
528 reader.project(&[]).expect("projects");
529 let chunk = reader.next_chunk().expect("reads").expect("a chunk");
530 assert_eq!(chunk.len(), 3);
531 assert_eq!(chunk.width(), 0);
532 }
533
534 #[test]
535 fn a_pipe_separated_file_reads_as_one() {
536 let mut reader = read("a|b\n1|x\n");
537 assert_eq!(reader.dialect().delimiter, b'|');
538 assert_eq!(all(&mut reader), [vec![Value::BigInt(1), Value::Varchar("x".into())]]);
539 }
540
541 #[test]
542 fn a_quoted_field_with_a_delimiter_in_it_is_one_value() {
543 let mut reader = read("a,b\n1,\"x,y\"\n");
544 assert_eq!(all(&mut reader), [vec![Value::BigInt(1), Value::Varchar("x,y".into())]]);
545 }
546
547 #[test]
548 fn more_rows_than_fit_one_chunk_arrive_as_more_than_one_chunk() {
549 let mut text = String::from("a\n");
550 for row in 0..VECTOR_SIZE + 5 {
551 text.push_str(&format!("{row}\n"));
552 }
553 let mut reader = read(&text);
554 let first = reader.next_chunk().expect("reads").expect("a chunk");
555 assert_eq!(first.len(), VECTOR_SIZE);
556 let second = reader.next_chunk().expect("reads").expect("a second chunk");
557 assert_eq!(second.len(), 5);
558 assert!(reader.next_chunk().expect("reads").is_none());
559 }
560
561 #[test]
562 fn a_value_the_sniffer_never_saw_is_an_error_rather_than_a_wider_column() {
563 let mut text = String::from("c\n");
567 for row in 0..infer::SAMPLE {
568 text.push_str(&format!("{row}\n"));
569 }
570 text.push_str("oops\n");
571 let mut reader = read(&text);
572 assert_eq!(reader.fields()[0].ty, LogicalType::BigInt);
573 let error = all_or_error(&mut reader).unwrap_err();
574 let line = infer::SAMPLE + 2;
575 assert!(error.message().starts_with(&format!("CSV Error on Line: {line}")), "{error}");
576 assert!(
577 error.message().contains("Could not convert string \"oops\" to 'BIGINT'"),
578 "{error}"
579 );
580 assert!(error.message().contains("sample_size = 20480"), "{error}");
581 }
582
583 #[test]
586 fn a_file_told_it_has_no_header_reads_its_first_line_as_a_row() {
587 let mut reader =
588 read_with("a,b\n1,x\n2,y\n", Given { header: Some(false), ..Given::default() });
589 assert_eq!(
590 names_and_types(&reader),
591 [
592 ("column0".to_string(), "VARCHAR".to_string()),
593 ("column1".to_string(), "VARCHAR".to_string()),
594 ]
595 );
596 assert_eq!(all(&mut reader).len(), 3);
597 }
598
599 #[test]
601 fn a_file_told_it_has_a_header_takes_its_first_line_as_the_names() {
602 let reader = read_with("1,2\n3,4\n", Given { header: Some(true), ..Given::default() });
603 assert_eq!(reader.fields().iter().map(|f| f.name.clone()).collect::<Vec<_>>(), ["1", "2"]);
604 }
605
606 #[test]
608 fn a_given_delimiter_is_the_delimiter_whatever_the_file_looks_like() {
609 let reader = read_with("a,b\n1,x\n", Given { delimiter: Some(b';'), ..Given::default() });
610 assert_eq!(reader.dialect().delimiter, b';');
611 assert_eq!(reader.fields().len(), 1);
612 }
613
614 #[test]
616 fn a_given_quote_makes_a_field_that_holds_the_delimiter_one_value() {
617 let mut reader =
618 read_with("a,b\n1,'x,y'\n", Given { quote: Some(b'\''), ..Given::default() });
619 assert_eq!(all(&mut reader), [vec![Value::BigInt(1), Value::Varchar("x,y".into())]]);
620 }
621
622 #[test]
624 fn the_block_says_set_by_user_for_what_the_call_gave_it() {
625 let mut text = String::from("c;d\n");
626 for row in 0..infer::SAMPLE {
627 text.push_str(&format!("{row};x\n"));
628 }
629 text.push_str("oops;x\n");
630 let given = Given { delimiter: Some(b';'), ..Given::default() };
631 let mut reader = read_with(&text, given);
632 let error = all_or_error(&mut reader).unwrap_err();
633 assert!(error.message().contains("delimiter = ; (Set By User)"), "{error}");
634 assert!(error.message().contains("header = true (Auto-Detected)"), "{error}");
635 }
636
637 #[test]
638 fn a_file_told_a_wider_type_than_it_sniffed_reads_its_whole_numbers_as_that_type() {
639 let mut reader = read("a\n1\n2\n");
643 assert_eq!(reader.fields()[0].ty, LogicalType::BigInt);
644 reader.retype(&[LogicalType::Double]).expect("one type for one column");
645 assert_eq!(reader.fields()[0].ty, LogicalType::Double);
646 assert_eq!(all(&mut reader), [[Value::Double(1.0)], [Value::Double(2.0)]]);
647 }
648
649 #[test]
650 fn a_type_list_that_is_not_as_long_as_the_projection_is_refused() {
651 let mut reader = read("a,b\n1,two\n");
652 let error = reader.retype(&[LogicalType::Double]).unwrap_err();
653 assert!(error.message().contains("1 types for a projection of 2 columns"), "{error}");
654 }
655
656 fn all_or_error(reader: &mut Reader) -> Result<Vec<Vec<Value>>> {
657 let mut rows = Vec::new();
658 while let Some(chunk) = reader.next_chunk()? {
659 for row in 0..chunk.len() {
660 rows.push((0..chunk.width()).map(|at| chunk.value_at(row, at)).collect());
661 }
662 }
663 Ok(rows)
664 }
665}