zql_cli/db/

dataset.rs

use crate::core::case::Payload;
use crate::db::format::Format;
use crate::db::layout::flatten::{FlattenLayout, FlattenVerbose};
use crate::db::layout::table::{TableHeader, TableLayout};
use crate::db::metadata::Metadata;
use crate::error::MyResult;
use crate::util::convert::format_null;
use crate::util::painter::{Painter, Style};
use csv::{Reader, ReaderBuilder};
use odbc_api::buffers::TextRowSet;
use odbc_api::handles::StatementImpl;
use odbc_api::{Cursor, CursorImpl, DataType, Nullability};
use std::borrow::Cow;
use std::cell::RefCell;
use std::cmp::Ordering;
use std::io::Read;
use std::iter::zip;

pub type Column = (String, Format, Metadata);
pub type Record = Vec<(String, Format)>;

enum Source<'a, R: Read> {
    Cursor(CursorImpl<StatementImpl<'a>>),
    File(Reader<R>),
    None,
}

pub struct Dataset<'a, R: Read> {
    pub columns: Vec<Column>,
    records: RefCell<Vec<(Record, bool)>>,
    source: RefCell<Source<'a, R>>,
    widths: RefCell<Option<Vec<Format>>>,
    auto: bool,
}

impl<'a, R: Read> Dataset<'a, R> {
    pub fn from_cursor(mut cursor: CursorImpl<StatementImpl<'a>>) -> MyResult<Option<Self>> {
        let columns = fetch_columns(&mut cursor)?;
        if !columns.is_empty() {
            let records = RefCell::new(Vec::new());
            let source = RefCell::new(Source::Cursor(cursor));
            let widths = RefCell::new(None);
            let dataset = Self { columns, records, source, widths, auto: false };
            Ok(Some(dataset))
        } else {
            Ok(None)
        }
    }

    pub fn from_file(reader: R, auto: bool) -> MyResult<Option<Self>> {
        let mut reader = ReaderBuilder::new().from_reader(reader);
        let headers = reader.headers()?;
        if !headers.is_empty() {
            let dtype = if auto { DataType::Unknown } else { DataType::Char { length: None } };
            let columns = headers
                .iter()
                .map(|name| measure_text(name))
                .map(|(name, format)| (name, format, Metadata::new(dtype, Nullability::Unknown)))
                .collect();
            let records = RefCell::new(Vec::new());
            let source = RefCell::new(Source::File(reader));
            let widths = RefCell::new(None);
            let dataset = Self { columns, records, source, widths, auto };
            Ok(Some(dataset))
        } else {
            Ok(None)
        }
    }

    pub fn from_keywords(
        column: &str,
        records: Vec<&str>,
    ) -> Self {
        let (name, format) = measure_text(column);
        let columns = vec![
            (name, format, Metadata::new(DataType::Unknown, Nullability::Unknown)),
        ];
        let records = records
            .into_iter()
            .map(|value| measure_text(value))
            .map(|pair| (vec![pair], false))
            .collect();
        let records = RefCell::new(records);
        let source = RefCell::new(Source::None);
        let widths = RefCell::new(None);
        Self { columns, records, source, widths, auto: false }
    }

    // noinspection DuplicatedCode
    pub fn from_tables(
        column1: Option<&str>,
        column2: Option<&str>,
        column3: &str,
        records: Vec<(&str, &str, &str)>,
    ) -> Self {
        let mut columns = Vec::new();
        let mut append = |column: &str, dtype: DataType| {
            let (name, format) = measure_text(column);
            columns.push((name, format, Metadata::new(dtype, Nullability::Unknown)));
        };
        let records = if let Some(column1) = column1 {
            if let Some(column2) = column2 {
                append(column1, DataType::Unknown);
                append(column2, DataType::Unknown);
                append(column3, DataType::Unknown);
                records
                    .into_iter()
                    .map(|(v1, v2, v3)| (measure_text(v1), measure_text(v2), measure_text(v3)))
                    .map(|(p1, p2, p3)| (vec![p1, p2, p3], false))
                    .collect()
            } else {
                append(column1, DataType::Unknown);
                append(column3, DataType::Unknown);
                records
                    .into_iter()
                    .map(|(v1, _, v3)| (measure_text(v1), measure_text(v3)))
                    .map(|(p1, p3)| (vec![p1, p3], false))
                    .collect()
            }
        } else {
            append(column3, DataType::Unknown);
            records
                .into_iter()
                .map(|(_, _, v3)| measure_text(v3))
                .map(|p3| (vec![p3], false))
                .collect()
        };
        let records = RefCell::new(records);
        let source = RefCell::new(Source::None);
        let widths = RefCell::new(None);
        Self { columns, records, source, widths, auto: false }
    }

    // noinspection DuplicatedCode
    pub fn from_columns(
        column1: Option<&str>,
        column2: Option<&str>,
        records: Vec<(&str, &str, &str, &str, Option<&Payload>)>,
    ) -> Self {
        let mut columns = Vec::new();
        let mut append = |column: &str, dtype: DataType| {
            let (name, format) = measure_text(column);
            columns.push((name, format, Metadata::new(dtype, Nullability::Unknown)));
        };
        let metadata = Metadata::new(DataType::Integer, Nullability::Unknown);
        let records = if let Some(column1) = column1 {
            if let Some(column2) = column2 {
                append(column1, DataType::Unknown);
                append(column2, DataType::Unknown);
                append("table", DataType::Unknown);
                append("column", DataType::Unknown);
                append("type", DataType::Unknown);
                append("nullable", DataType::Unknown);
                append("index", DataType::Integer);
                records
                    .into_iter()
                    .map(|(v1, v2, v3, v4, p5)| (
                        measure_text(v1),
                        measure_text(v2),
                        measure_text(v3),
                        measure_text(v4),
                        measure_type(p5),
                        measure_null(p5),
                        measure_index(p5, &metadata),
                        group_index(p5),
                    ))
                    .map(|(p1, p2, p3, p4, p5, p6, p7, sep)| (vec![p1, p2, p3, p4, p5, p6, p7], sep))
                    .collect()
            } else {
                append(column1, DataType::Unknown);
                append("table", DataType::Unknown);
                append("column", DataType::Unknown);
                append("type", DataType::Unknown);
                append("nullable", DataType::Unknown);
                append("index", DataType::Integer);
                records
                    .into_iter()
                    .map(|(v1, _, v3, v4, p5)| (
                        measure_text(v1),
                        measure_text(v3),
                        measure_text(v4),
                        measure_type(p5),
                        measure_null(p5),
                        measure_index(p5, &metadata),
                        group_index(p5),
                    ))
                    .map(|(p1, p3, p4, p5, p6, p7, sep)| (vec![p1, p3, p4, p5, p6, p7], sep))
                    .collect()
            }
        } else {
            append("table", DataType::Unknown);
            append("column", DataType::Unknown);
            append("type", DataType::Unknown);
            append("nullable", DataType::Unknown);
            append("index", DataType::Integer);
            records
                .into_iter()
                .map(|(_, _, v3, v4, p5)| (
                    measure_text(v3),
                    measure_text(v4),
                    measure_type(p5),
                    measure_null(p5),
                    measure_index(p5, &metadata),
                    group_index(p5),
                ))
                .map(|(p3, p4, p5, p6, p7, sep)| (vec![p3, p4, p5, p6, p7], sep))
                .collect()
        };
        let records = RefCell::new(records);
        let source = RefCell::new(Source::None);
        let widths = RefCell::new(None);
        Self { columns, records, source, widths, auto: false }
    }

    pub fn get_columns(&self, painter: Painter) -> Vec<String> {
        self.columns
            .iter()
            .map(|(x, _, _)| painter.wrap_style(x, Style::Column))
            .collect()
    }

    pub fn has_records(&self) -> bool {
        // Must be called after records have been retrieved.
        !self.records.borrow().is_empty()
    }

    pub fn group_records(&mut self, group: usize) {
        // Must be called after records have been retrieved.  Note, this
        // relies on the sort algorithm being stable, i.e. not reordering
        // equal elements.
        self.records.borrow_mut().sort_by(|(l, _), (r, _)| Self::compare_records(l, r, group));
    }

    fn compare_records(left: &Record, right: &Record, group: usize) -> Ordering {
        let left = left.get(group).map(|(x, _)| x.as_str()).unwrap_or_default();
        let right = right.get(group).map(|(x, _)| x.as_str()).unwrap_or_default();
        left.cmp(right)
    }

    pub fn get_records<F>(&self, limit: usize, mut function: F) -> MyResult<()> where
        F: FnMut(Vec<Cow<str>>) -> MyResult<()>,
    {
        match self.source.replace(Source::None) {
            Source::Cursor(cursor) => {
                fetch_cursor(cursor, limit, function)
            }
            Source::File(reader) => {
                fetch_file(reader, function)
            }
            Source::None => {
                let records = self.records.borrow();
                for (record, _) in records.iter() {
                    let record = record.iter().map(|(x, _)| Cow::Borrowed(x.as_ref())).collect();
                    function(record)?;
                }
                Ok(())
            }
        }
    }

    pub fn get_measured<F>(&self, limit: usize, mut function: F) -> MyResult<()> where
        F: FnMut(Record, bool) -> MyResult<()>,
    {
        match self.source.replace(Source::None) {
            Source::Cursor(cursor) => {
                fetch_cursor(cursor, limit, |record| {
                    let record = zip(record, &self.columns)
                        .map(|(value, (_, _, metadata))| measure_value(value, metadata))
                        .collect();
                    function(record, false)?;
                    Ok(())
                })
            }
            Source::File(reader) => {
                fetch_file(reader, |record| {
                    let record = record
                        .into_iter()
                        .map(measure_text)
                        .collect();
                    function(record, false)?;
                    Ok(())
                })
            }
            Source::None => {
                let mut records = self.records.borrow_mut();
                let mut first = true;
                for (record, sep) in records.drain(..) {
                    function(record, sep && !first)?;
                    first = false;
                }
                Ok(())
            }
        }
    }

    pub fn store_records(&mut self) -> MyResult<()> {
        match self.source.replace(Source::None) {
            Source::Cursor(cursor) => {
                fetch_cursor(cursor, 4096, |record| {
                    self.push_record(record);
                    Ok(())
                })
            }
            Source::File(reader) => {
                fetch_file(reader, |record| {
                    self.push_record(record);
                    Ok(())
                })
            }
            Source::None => {
                Ok(())
            }
        }
    }

    pub fn push_record(&mut self, record: Vec<Cow<str>>) {
        let record = zip(record, &self.columns)
            .map(|(value, (_, _, metadata))| measure_value(value, metadata))
            .collect::<Vec<_>>();
        self.records.borrow_mut().push((record, false));
    }

    pub fn get_widths(&self) -> Vec<Format> {
        self.widths.borrow_mut().get_or_insert_with(|| {
            // If we're formatting data from a CSV file, we do not know
            // in advance whether each column will contain text or numbers,
            // so we create the column metadata with no format template.
            // As we read the file, we parse each value via a regex, and
            // override the template on the first text value.  Thereafter,
            // each value in that column will be parsed as text.
            if self.auto {
                let records = self.records.borrow_mut().drain(..)
                    .map(|(record, sep)| (self.do_reformat(record), sep))
                    .collect();
                self.records.replace(records);
            }
            // Get the maximum text or numeric width over every value in
            // each column.  At this stage, each column should contain
            // consistently formatted values.  The resulting widths will
            // optionally be compared against the column header widths.
            let mut widths = [Format::Text(0)].repeat(self.columns.len());
            for (record, _) in self.records.borrow().iter() {
                widths = zip(record, &widths)
                    .map(|((_, format), width)| Format::max(format, width))
                    .collect();
            }
            widths
        }).clone()
    }

    fn do_reformat(&self, record: Record) -> Record {
        // For each value in the record, test whether the value is numeric,
        // and the template in the column metadata is text.  If this is
        // true, force the value format to text.
        zip(record, &self.columns)
            .map(|((value, format), (_, _, metadata))| (value, metadata.do_reformat(format)))
            .collect()
    }

    pub fn measure_table(&self) -> usize {
        let widths = self.get_widths();
        let total = zip(&self.columns, &widths)
            .map(|((_, format, _), width)| Format::max(format, width))
            .map(|format| format.total())
            .sum::<usize>();
        total + (widths.len() * 2)
    }

    pub fn adjust_header(&self, header: TableHeader, interact: bool) -> TableHeader {
        if interact || self.columns.len() >= 2 { header } else { TableHeader::Simple }
    }

    pub fn into_table(self, header: TableHeader, group: Option<&str>) -> TableLayout<'a, R> {
        TableLayout::from_dataset(self, header, group)
    }

    pub fn into_flatten(self, verbose: FlattenVerbose) -> FlattenLayout<'a, R> {
        FlattenLayout::from_dataset(self, verbose)
    }
}

fn fetch_columns<C: Cursor>(cursor: &mut C) -> MyResult<Vec<Column>> {
    let size = cursor.num_result_cols()? as u16;
    let mut columns = Vec::with_capacity(size as usize);
    for index in 1..=size {
        let name = cursor.col_name(index)?;
        let dtype = cursor.col_data_type(index)?;
        let null = cursor.col_nullability(index)?;
        let (name, format) = measure_text(name);
        let metadata = Metadata::new(dtype, null);
        columns.push((name, format, metadata));
    }
    Ok(columns)
}

fn fetch_cursor<C, F>(
    mut cursor: C,
    limit: usize,
    mut function: F,
) -> MyResult<()> where
    C: Cursor,
    F: FnMut(Vec<Cow<str>>) -> MyResult<()>,
{
    let rowset = TextRowSet::for_cursor(5000, &mut cursor, Some(limit))?;
    let mut cursor = cursor.bind_buffer(rowset)?;
    while let Some(batch) = cursor.fetch()? {
        let rows = batch.num_rows();
        let cols = batch.num_cols();
        for row in 0..rows {
            let record = (0..cols)
                .map(|col| fetch_value(batch, col, row))
                .collect();
            function(record)?;
        }
    }
    Ok(())
}

fn fetch_value(
    batch: &TextRowSet,
    col: usize,
    row: usize,
) -> Cow<'_, str> {
    let value = batch
        .at_as_str(col, row)
        .unwrap_or_default()
        .unwrap_or_default();
    Cow::Borrowed(value)
}

fn fetch_file<R, F>(
    mut reader: Reader<R>,
    mut function: F,
) -> MyResult<()> where
    R: Read,
    F: FnMut(Vec<Cow<str>>) -> MyResult<()>,
{
    for record in reader.records() {
        let record = record?;
        let record = record.iter().map(Cow::from).collect();
        function(record)?;
    }
    Ok(())
}

fn measure_text<V: Into<String>>(value: V) -> (String, Format) {
    let value = value.into();
    let length = value.chars().count();
    let format = Format::Text(length);
    (value, format)
}

fn measure_value<V: Into<String>>(value: V, metadata: &Metadata) -> (String, Format) {
    let value = value.into();
    let length = value.chars().count();
    let format = metadata.measure_value(&value, length);
    (value, format)
}

fn measure_type(payload: Option<&Payload>) -> (String, Format) {
    let dtype = payload.map(|p| p.dtype.as_str()).unwrap_or_default();
    measure_text(dtype)
}

fn measure_null(payload: Option<&Payload>) -> (String, Format) {
    let null = payload.map(|p| p.null).unwrap_or_default();
    measure_text(format_null(&null))
}

fn measure_index(payload: Option<&Payload>, metadata: &Metadata) -> (String, Format) {
    let index = payload.map(|p| p.index).unwrap_or_default();
    measure_value(index.to_string(), metadata)
}

fn group_index(payload: Option<&Payload>) -> bool {
    let index = payload.map(|p| p.index).unwrap_or_default();
    index == 1
}

#[cfg(test)]
mod tests {
    use crate::core::case::Payload;
    use crate::db::dataset::{Dataset, Record};
    use crate::{cow_str, str_vec};
    use odbc_api::Nullability::{NoNulls, Nullable};
    use pretty_assertions::assert_eq;
    use std::borrow::Cow;
    use std::fs::File;

    #[test]
    fn test_dataset_is_created_from_keywords() {
        let dataset = Dataset::<File>::from_keywords("keyword", vec![
            "ADD",
            "ALTER",
            "AND",
        ]);
        assert_eq!(get_columns(&dataset), str_vec![
            "keyword",
        ]);
        assert_eq!(get_records(&dataset), vec![
            str_vec!["ADD"],
            str_vec!["ALTER"],
            str_vec!["AND"],
        ]);
    }

    #[test]
    fn test_dataset_is_created_from_tables_with_database_and_schema() {
        let dataset = Dataset::<File>::from_tables(Some("database"), Some("schema"), "table", vec![
            ("Archive", "Main", "Article"),
            ("Archive", "Main", "Journal"),
            ("Business", "Admin", "Server"),
            ("Business", "Main", "Account"),
            ("Business", "Main", "Person"),
        ]);
        assert_eq!(get_columns(&dataset), str_vec![
            "database",
            "schema",
            "table",
        ]);
        assert_eq!(get_records(&dataset), vec![
            str_vec!["Archive", "Main", "Article"],
            str_vec!["Archive", "Main", "Journal"],
            str_vec!["Business", "Admin", "Server"],
            str_vec!["Business", "Main", "Account"],
            str_vec!["Business", "Main", "Person"],
        ]);
    }

    #[test]
    fn test_dataset_is_created_from_tables_with_database_no_schema() {
        let dataset = Dataset::<File>::from_tables(Some("database"), None, "table", vec![
            ("Archive", "", "Article"),
            ("Archive", "", "Journal"),
            ("Business", "", "Account"),
            ("Business", "", "Person"),
        ]);
        assert_eq!(get_columns(&dataset), str_vec![
            "database",
            "table",
        ]);
        assert_eq!(get_records(&dataset), vec![
            str_vec!["Archive", "Article"],
            str_vec!["Archive", "Journal"],
            str_vec!["Business", "Account"],
            str_vec!["Business", "Person"],
        ]);
    }

    #[test]
    fn test_dataset_is_created_from_tables_no_database_or_schema() {
        let dataset = Dataset::<File>::from_tables(None, None, "table", vec![
            ("", "", "Account"),
            ("", "", "Person"),
        ]);
        assert_eq!(get_columns(&dataset), str_vec![
            "table",
        ]);
        assert_eq!(get_records(&dataset), vec![
            str_vec!["Account"],
            str_vec!["Person"],
        ]);
    }

    #[test]
    fn test_dataset_is_created_from_columns_with_database_and_schema() {
        let dataset = Dataset::<File>::from_columns(Some("database"), Some("schema"), vec![
            ("Archive", "Main", "Article", "Journal", Some(&Payload::new(cow_str!("INTEGER"), NoNulls, 1))),
            ("Archive", "Main", "Article", "Date", Some(&Payload::new(cow_str!("DATE"), NoNulls, 2))),
            ("Archive", "Main", "Article", "Author", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 3))),
            ("Archive", "Main", "Article", "Text", Some(&Payload::new(cow_str!("VARCHAR(4000)"), NoNulls, 4))),
            ("Archive", "Main", "Article", "Appendix", Some(&Payload::new(cow_str!("VARCHAR(4000)"), Nullable, 5))),
            ("Archive", "Main", "Journal", "Country", Some(&Payload::new(cow_str!("INTEGER"), NoNulls, 1))),
            ("Archive", "Main", "Journal", "Price", Some(&Payload::new(cow_str!("DECIMAL(10, 2)"), NoNulls, 2))),
            ("Archive", "Main", "Journal", "Available", Some(&Payload::new(cow_str!("BIT"), NoNulls, 3))),
            ("Business", "Admin", "Server", "Name", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 1))),
            ("Business", "Admin", "Server", "Location", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 2))),
            ("Business", "Admin", "Server", "Hardware", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 3))),
            ("Business", "Main", "Account", "Company", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 1))),
            ("Business", "Main", "Account", "Branch", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 2))),
            ("Business", "Main", "Account", "Amount", Some(&Payload::new(cow_str!("DECIMAL(10, 2)"), NoNulls, 3))),
            ("Business", "Main", "Account", "Active", Some(&Payload::new(cow_str!("BIT"), NoNulls, 4))),
            ("Business", "Main", "Person", "Forename", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 1))),
            ("Business", "Main", "Person", "Surname", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 2))),
            ("Business", "Main", "Person", "Address", Some(&Payload::new(cow_str!("VARCHAR(20)"), Nullable, 3))),
            ("Business", "Main", "Person", "Birthday", Some(&Payload::new(cow_str!("DATE"), Nullable, 4))),
        ]);
        assert_eq!(get_columns(&dataset), str_vec![
            "database",
            "schema",
            "table",
            "column",
            "type",
            "nullable",
            "index",
        ]);
        assert_eq!(get_records(&dataset), vec![
            str_vec!["Archive", "Main", "Article", "Journal", "INTEGER", "NOT NULL", "1"],
            str_vec!["Archive", "Main", "Article", "Date", "DATE", "NOT NULL", "2"],
            str_vec!["Archive", "Main", "Article", "Author", "VARCHAR(20)", "NOT NULL", "3"],
            str_vec!["Archive", "Main", "Article", "Text", "VARCHAR(4000)", "NOT NULL", "4"],
            str_vec!["Archive", "Main", "Article", "Appendix", "VARCHAR(4000)", "NULL", "5"],
            str_vec!["Archive", "Main", "Journal", "Country", "INTEGER", "NOT NULL", "1"],
            str_vec!["Archive", "Main", "Journal", "Price", "DECIMAL(10, 2)", "NOT NULL", "2"],
            str_vec!["Archive", "Main", "Journal", "Available", "BIT", "NOT NULL", "3"],
            str_vec!["Business", "Admin", "Server", "Name", "VARCHAR(20)", "NOT NULL", "1"],
            str_vec!["Business", "Admin", "Server", "Location", "VARCHAR(20)", "NOT NULL", "2"],
            str_vec!["Business", "Admin", "Server", "Hardware", "VARCHAR(20)", "NOT NULL", "3"],
            str_vec!["Business", "Main", "Account", "Company", "VARCHAR(20)", "NOT NULL", "1"],
            str_vec!["Business", "Main", "Account", "Branch", "VARCHAR(20)", "NOT NULL", "2"],
            str_vec!["Business", "Main", "Account", "Amount", "DECIMAL(10, 2)", "NOT NULL", "3"],
            str_vec!["Business", "Main", "Account", "Active", "BIT", "NOT NULL", "4"],
            str_vec!["Business", "Main", "Person", "Forename", "VARCHAR(20)", "NOT NULL", "1"],
            str_vec!["Business", "Main", "Person", "Surname", "VARCHAR(20)", "NOT NULL", "2"],
            str_vec!["Business", "Main", "Person", "Address", "VARCHAR(20)", "NULL", "3"],
            str_vec!["Business", "Main", "Person", "Birthday", "DATE", "NULL", "4"],
        ]);
    }

    #[test]
    fn test_dataset_is_created_from_columns_with_database_no_schema() {
        let dataset = Dataset::<File>::from_columns(Some("database"), None, vec![
            ("Archive", "", "Article", "Journal", Some(&Payload::new(cow_str!("INTEGER"), NoNulls, 1))),
            ("Archive", "", "Article", "Date", Some(&Payload::new(cow_str!("DATE"), NoNulls, 2))),
            ("Archive", "", "Article", "Author", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 3))),
            ("Archive", "", "Article", "Text", Some(&Payload::new(cow_str!("VARCHAR(4000)"), NoNulls, 4))),
            ("Archive", "", "Article", "Appendix", Some(&Payload::new(cow_str!("VARCHAR(4000)"), Nullable, 5))),
            ("Archive", "", "Journal", "Country", Some(&Payload::new(cow_str!("INTEGER"), NoNulls, 1))),
            ("Archive", "", "Journal", "Price", Some(&Payload::new(cow_str!("DECIMAL(10, 2)"), NoNulls, 2))),
            ("Archive", "", "Journal", "Available", Some(&Payload::new(cow_str!("BIT"), NoNulls, 3))),
            ("Business", "", "Account", "Company", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 1))),
            ("Business", "", "Account", "Branch", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 2))),
            ("Business", "", "Account", "Amount", Some(&Payload::new(cow_str!("DECIMAL(10, 2)"), NoNulls, 3))),
            ("Business", "", "Account", "Active", Some(&Payload::new(cow_str!("BIT"), NoNulls, 4))),
            ("Business", "", "Person", "Forename", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 1))),
            ("Business", "", "Person", "Surname", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 2))),
            ("Business", "", "Person", "Address", Some(&Payload::new(cow_str!("VARCHAR(20)"), Nullable, 3))),
            ("Business", "", "Person", "Birthday", Some(&Payload::new(cow_str!("DATE"), Nullable, 4))),
        ]);
        assert_eq!(get_columns(&dataset), str_vec![
            "database",
            "table",
            "column",
            "type",
            "nullable",
            "index",
        ]);
        assert_eq!(get_records(&dataset), vec![
            str_vec!["Archive", "Article", "Journal", "INTEGER", "NOT NULL", "1"],
            str_vec!["Archive", "Article", "Date", "DATE", "NOT NULL", "2"],
            str_vec!["Archive", "Article", "Author", "VARCHAR(20)", "NOT NULL", "3"],
            str_vec!["Archive", "Article", "Text", "VARCHAR(4000)", "NOT NULL", "4"],
            str_vec!["Archive", "Article", "Appendix", "VARCHAR(4000)", "NULL", "5"],
            str_vec!["Archive", "Journal", "Country", "INTEGER", "NOT NULL", "1"],
            str_vec!["Archive", "Journal", "Price", "DECIMAL(10, 2)", "NOT NULL", "2"],
            str_vec!["Archive", "Journal", "Available", "BIT", "NOT NULL", "3"],
            str_vec!["Business", "Account", "Company", "VARCHAR(20)", "NOT NULL", "1"],
            str_vec!["Business", "Account", "Branch", "VARCHAR(20)", "NOT NULL", "2"],
            str_vec!["Business", "Account", "Amount", "DECIMAL(10, 2)", "NOT NULL", "3"],
            str_vec!["Business", "Account", "Active", "BIT", "NOT NULL", "4"],
            str_vec!["Business", "Person", "Forename", "VARCHAR(20)", "NOT NULL", "1"],
            str_vec!["Business", "Person", "Surname", "VARCHAR(20)", "NOT NULL", "2"],
            str_vec!["Business", "Person", "Address", "VARCHAR(20)", "NULL", "3"],
            str_vec!["Business", "Person", "Birthday", "DATE", "NULL", "4"],
        ]);
    }

    #[test]
    fn test_dataset_is_created_from_columns_no_database_or_schema() {
        let dataset = Dataset::<File>::from_columns(None, None, vec![
            ("", "", "Account", "Company", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 1))),
            ("", "", "Account", "Branch", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 2))),
            ("", "", "Account", "Amount", Some(&Payload::new(cow_str!("DECIMAL(10, 2)"), NoNulls, 3))),
            ("", "", "Account", "Active", Some(&Payload::new(cow_str!("BIT"), NoNulls, 4))),
            ("", "", "Person", "Forename", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 1))),
            ("", "", "Person", "Surname", Some(&Payload::new(cow_str!("VARCHAR(20)"), NoNulls, 2))),
            ("", "", "Person", "Address", Some(&Payload::new(cow_str!("VARCHAR(20)"), Nullable, 3))),
            ("", "", "Person", "Birthday", Some(&Payload::new(cow_str!("DATE"), Nullable, 4))),
        ]);
        assert_eq!(get_columns(&dataset), str_vec![
            "table",
            "column",
            "type",
            "nullable",
            "index",
        ]);
        assert_eq!(get_records(&dataset), vec![
            str_vec!["Account", "Company", "VARCHAR(20)", "NOT NULL", "1"],
            str_vec!["Account", "Branch", "VARCHAR(20)", "NOT NULL", "2"],
            str_vec!["Account", "Amount", "DECIMAL(10, 2)", "NOT NULL", "3"],
            str_vec!["Account", "Active", "BIT", "NOT NULL", "4"],
            str_vec!["Person", "Forename", "VARCHAR(20)", "NOT NULL", "1"],
            str_vec!["Person", "Surname", "VARCHAR(20)", "NOT NULL", "2"],
            str_vec!["Person", "Address", "VARCHAR(20)", "NULL", "3"],
            str_vec!["Person", "Birthday", "DATE", "NULL", "4"],
        ]);
    }

    fn get_columns(dataset: &Dataset<File>) -> Vec<String> {
        dataset.columns.iter().map(|(x, _, _)| x.clone()).collect()
    }

    fn get_records(dataset: &Dataset<File>) -> Vec<Vec<String>> {
        dataset.records.borrow().iter().map(|(x, _)| get_values(x)).collect()
    }

    fn get_values(record: &Record) -> Vec<String> {
        record.iter().map(|(x, _)| x.clone()).collect()
    }
}