Database commands (#5307)

* database commands

* db commands

* filesystem opens sqlite file

* clippy error

* corrected error in ci file

* removes matrix flag from ci

* flax matrix for clippy

* add conditional compile for tests

* add conditional compile for tests

* correct order of command

* correct error msg

* correct typo

crates/nu-command/src/database/commands/collect.rs

@@ -0,0 +1,49 @@
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, Example, IntoPipelineData, PipelineData, ShellError, Signature,
+};
+
+use super::super::SQLiteDatabase;
+
+#[derive(Clone)]
+pub struct CollectDb;
+
+impl Command for CollectDb {
+    fn name(&self) -> &str {
+        "db collect"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name()).category(Category::Custom("database".into()))
+    }
+
+    fn usage(&self) -> &str {
+        "Query a database using SQL."
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![Example {
+            description: "Collect from a select query",
+            example: "open foo.db | db select a | db from table_1 | db collect",
+            result: None,
+        }]
+    }
+
+    fn search_terms(&self) -> Vec<&str> {
+        vec!["database", "collect"]
+    }
+
+    fn run(
+        &self,
+        _engine_state: &EngineState,
+        _stack: &mut Stack,
+        call: &Call,
+        input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        let db = SQLiteDatabase::try_from_pipeline(input, call.head)?;
+
+        db.collect(call.head)
+            .map(IntoPipelineData::into_pipeline_data)
+    }
+}

crates/nu-command/src/database/commands/command.rs

@@ -0,0 +1,42 @@
+use nu_engine::get_full_help;
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, IntoPipelineData, PipelineData, ShellError, Signature, Value,
+};
+
+#[derive(Clone)]
+pub struct Database;
+
+impl Command for Database {
+    fn name(&self) -> &str {
+        "db"
+    }
+
+    fn usage(&self) -> &str {
+        "Database commands"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name()).category(Category::Custom("database".into()))
+    }
+
+    fn run(
+        &self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+        call: &Call,
+        _input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        Ok(Value::String {
+            val: get_full_help(
+                &Database.signature(),
+                &Database.examples(),
+                engine_state,
+                stack,
+            ),
+            span: call.head,
+        }
+        .into_pipeline_data())
+    }
+}

crates/nu-command/src/database/commands/describe.rs

@@ -0,0 +1,47 @@
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, Example, IntoPipelineData, PipelineData, ShellError, Signature,
+};
+
+use super::super::SQLiteDatabase;
+
+#[derive(Clone)]
+pub struct DescribeDb;
+
+impl Command for DescribeDb {
+    fn name(&self) -> &str {
+        "db describe"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name()).category(Category::Custom("database".into()))
+    }
+
+    fn usage(&self) -> &str {
+        "Describes connection and query of the DB object"
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![Example {
+            description: "Describe SQLite database constructed query",
+            example: "db open foo.db | db select table_1 | db describe",
+            result: None,
+        }]
+    }
+
+    fn search_terms(&self) -> Vec<&str> {
+        vec!["database", "SQLite"]
+    }
+
+    fn run(
+        &self,
+        _engine_state: &EngineState,
+        _stack: &mut Stack,
+        call: &Call,
+        input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        let db = SQLiteDatabase::try_from_pipeline(input, call.head)?;
+        Ok(db.describe(call.head).into_pipeline_data())
+    }
+}

crates/nu-command/src/database/commands/from.rs

@@ -0,0 +1,130 @@
+use super::super::SQLiteDatabase;
+use nu_engine::CallExt;
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, Example, IntoPipelineData, PipelineData, ShellError, Signature, SyntaxShape,
+};
+use sqlparser::ast::{Ident, ObjectName, Query, Select, SetExpr, TableFactor, TableWithJoins};
+
+#[derive(Clone)]
+pub struct FromDb;
+
+impl Command for FromDb {
+    fn name(&self) -> &str {
+        "db from"
+    }
+
+    fn usage(&self) -> &str {
+        "Select section from query statement for a DB"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name())
+            .required(
+                "select",
+                SyntaxShape::String,
+                "Name of table to select from",
+            )
+            .category(Category::Custom("database".into()))
+    }
+
+    fn search_terms(&self) -> Vec<&str> {
+        vec!["database", "from"]
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![Example {
+            description: "Selects table from database",
+            example: "db open db.mysql | db from table_a",
+            result: None,
+        }]
+    }
+
+    fn run(
+        &self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+        call: &Call,
+        input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        let table: String = call.req(engine_state, stack, 0)?;
+
+        let mut db = SQLiteDatabase::try_from_pipeline(input, call.head)?;
+        db.query = match db.query {
+            None => Some(create_query(table)),
+            Some(query) => Some(modify_query(query, table)),
+        };
+
+        Ok(db.into_value(call.head).into_pipeline_data())
+    }
+}
+
+fn create_query(table: String) -> Query {
+    Query {
+        with: None,
+        body: SetExpr::Select(Box::new(create_select(table))),
+        order_by: Vec::new(),
+        limit: None,
+        offset: None,
+        fetch: None,
+        lock: None,
+    }
+}
+
+fn modify_query(mut query: Query, table: String) -> Query {
+    query.body = match query.body {
+        SetExpr::Select(select) => SetExpr::Select(Box::new(modify_select(select, table))),
+        _ => SetExpr::Select(Box::new(create_select(table))),
+    };
+
+    query
+}
+
+fn modify_select(select: Box<Select>, table: String) -> Select {
+    Select {
+        from: create_from(table),
+        ..select.as_ref().clone()
+    }
+}
+
+fn create_select(table: String) -> Select {
+    Select {
+        distinct: false,
+        top: None,
+        projection: Vec::new(),
+        into: None,
+        from: create_from(table),
+        lateral_views: Vec::new(),
+        selection: None,
+        group_by: Vec::new(),
+        cluster_by: Vec::new(),
+        distribute_by: Vec::new(),
+        sort_by: Vec::new(),
+        having: None,
+    }
+}
+
+// This function needs more work
+// It needs to define multi tables and joins
+// I assume we will need to define expressions for the columns instead of strings
+fn create_from(table: String) -> Vec<TableWithJoins> {
+    let ident = Ident {
+        value: table,
+        quote_style: None,
+    };
+
+    let table_factor = TableFactor::Table {
+        name: ObjectName(vec![ident]),
+        alias: None,
+        args: Vec::new(),
+        with_hints: Vec::new(),
+    };
+
+    let table = TableWithJoins {
+        relation: table_factor,
+        joins: Vec::new(),
+    };
+
+    vec![table]
+}

crates/nu-command/src/database/commands/mod.rs

@@ -0,0 +1,31 @@
+mod collect;
+mod command;
+mod describe;
+mod from;
+mod open;
+mod query;
+mod select;
+mod utils;
+
+use collect::CollectDb;
+use command::Database;
+use describe::DescribeDb;
+use from::FromDb;
+use nu_protocol::engine::StateWorkingSet;
+use open::OpenDb;
+use query::QueryDb;
+use select::SelectDb;
+
+pub fn add_database_decls(working_set: &mut StateWorkingSet) {
+    macro_rules! bind_command {
+            ( $command:expr ) => {
+                working_set.add_decl(Box::new($command));
+            };
+            ( $( $command:expr ),* ) => {
+                $( working_set.add_decl(Box::new($command)); )*
+            };
+        }
+
+    // Series commands
+    bind_command!(CollectDb, Database, DescribeDb, FromDb, QueryDb, SelectDb, OpenDb);
+}

crates/nu-command/src/database/commands/open.rs

@@ -0,0 +1,52 @@
+use super::super::SQLiteDatabase;
+use nu_engine::CallExt;
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, Example, IntoPipelineData, PipelineData, ShellError, Signature, Spanned, SyntaxShape,
+};
+use std::path::PathBuf;
+
+#[derive(Clone)]
+pub struct OpenDb;
+
+impl Command for OpenDb {
+    fn name(&self) -> &str {
+        "db open"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name())
+            .required("query", SyntaxShape::Filepath, "SQLite file to be opened")
+            .category(Category::Custom("database".into()))
+    }
+
+    fn usage(&self) -> &str {
+        "Open a database"
+    }
+
+    fn search_terms(&self) -> Vec<&str> {
+        vec!["database", "open"]
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![Example {
+            description: "",
+            example: r#"""#,
+            result: None,
+        }]
+    }
+
+    fn run(
+        &self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+        call: &Call,
+        _input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        let path: Spanned<PathBuf> = call.req(engine_state, stack, 0)?;
+
+        SQLiteDatabase::try_from_path(path.item.as_path(), path.span)
+            .map(|db| db.into_value(call.head).into_pipeline_data())
+    }
+}

crates/nu-command/src/database/commands/query.rs

@@ -0,0 +1,57 @@
+use nu_engine::CallExt;
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, Example, IntoPipelineData, PipelineData, ShellError, Signature, Spanned, SyntaxShape,
+};
+
+use super::super::SQLiteDatabase;
+
+#[derive(Clone)]
+pub struct QueryDb;
+
+impl Command for QueryDb {
+    fn name(&self) -> &str {
+        "db query"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name())
+            .required(
+                "query",
+                SyntaxShape::String,
+                "SQL to execute against the database",
+            )
+            .category(Category::Custom("database".into()))
+    }
+
+    fn usage(&self) -> &str {
+        "Query a database using SQL."
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![Example {
+            description: "Get 1 table out of a SQLite database",
+            example: r#"db open foo.db | db query "SELECT * FROM Bar""#,
+            result: None,
+        }]
+    }
+
+    fn search_terms(&self) -> Vec<&str> {
+        vec!["database", "SQLite"]
+    }
+
+    fn run(
+        &self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+        call: &Call,
+        input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        let sql: Spanned<String> = call.req(engine_state, stack, 0)?;
+
+        let db = SQLiteDatabase::try_from_pipeline(input, call.head)?;
+        db.query(&sql, call.head)
+            .map(IntoPipelineData::into_pipeline_data)
+    }
+}

crates/nu-command/src/database/commands/select.rs

@@ -0,0 +1,131 @@
+use super::{super::SQLiteDatabase, utils::extract_strings};
+use nu_engine::CallExt;
+use nu_protocol::{
+    ast::Call,
+    engine::{Command, EngineState, Stack},
+    Category, Example, IntoPipelineData, PipelineData, ShellError, Signature, SyntaxShape, Value,
+};
+use sqlparser::ast::{Expr, Ident, Query, Select, SelectItem, SetExpr};
+
+#[derive(Clone)]
+pub struct SelectDb;
+
+impl Command for SelectDb {
+    fn name(&self) -> &str {
+        "db select"
+    }
+
+    fn usage(&self) -> &str {
+        "Creates a select statement for a DB"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name())
+            .required(
+                "select",
+                SyntaxShape::Any,
+                "Select expression(s) on the table",
+            )
+            .category(Category::Custom("database".into()))
+    }
+
+    fn search_terms(&self) -> Vec<&str> {
+        vec!["database", "select"]
+    }
+
+    fn examples(&self) -> Vec<Example> {
+        vec![
+            Example {
+                description: "selects a column from a database",
+                example: "db open db.mysql | db select a",
+                result: None,
+            },
+            Example {
+                description: "selects columns from a database",
+                example: "db open db.mysql | db select [a, b, c]",
+                result: None,
+            },
+        ]
+    }
+
+    fn run(
+        &self,
+        engine_state: &EngineState,
+        stack: &mut Stack,
+        call: &Call,
+        input: PipelineData,
+    ) -> Result<PipelineData, ShellError> {
+        let value: Value = call.req(engine_state, stack, 0)?;
+        let expressions = extract_strings(value)?;
+
+        let mut db = SQLiteDatabase::try_from_pipeline(input, call.head)?;
+        db.query = match db.query {
+            None => Some(create_query(expressions)),
+            Some(query) => Some(modify_query(query, expressions)),
+        };
+
+        Ok(db.into_value(call.head).into_pipeline_data())
+    }
+}
+
+fn create_query(expressions: Vec<String>) -> Query {
+    Query {
+        with: None,
+        body: SetExpr::Select(Box::new(create_select(expressions))),
+        order_by: Vec::new(),
+        limit: None,
+        offset: None,
+        fetch: None,
+        lock: None,
+    }
+}
+
+fn modify_query(mut query: Query, expressions: Vec<String>) -> Query {
+    query.body = match query.body {
+        SetExpr::Select(select) => SetExpr::Select(Box::new(modify_select(select, expressions))),
+        _ => SetExpr::Select(Box::new(create_select(expressions))),
+    };
+
+    query
+}
+
+fn modify_select(select: Box<Select>, expressions: Vec<String>) -> Select {
+    Select {
+        projection: create_projection(expressions),
+        ..select.as_ref().clone()
+    }
+}
+
+fn create_select(expressions: Vec<String>) -> Select {
+    Select {
+        distinct: false,
+        top: None,
+        projection: create_projection(expressions),
+        into: None,
+        from: Vec::new(),
+        lateral_views: Vec::new(),
+        selection: None,
+        group_by: Vec::new(),
+        cluster_by: Vec::new(),
+        distribute_by: Vec::new(),
+        sort_by: Vec::new(),
+        having: None,
+    }
+}
+
+// This function needs more work
+// It needs to define alias and functions in the columns
+// I assume we will need to define expressions for the columns instead of strings
+fn create_projection(expressions: Vec<String>) -> Vec<SelectItem> {
+    expressions
+        .into_iter()
+        .map(|expression| {
+            let expr = Expr::Identifier(Ident {
+                value: expression,
+                quote_style: None,
+            });
+
+            SelectItem::UnnamedExpr(expr)
+        })
+        .collect()
+}

crates/nu-command/src/database/commands/utils.rs

@@ -0,0 +1,15 @@
+use nu_protocol::{FromValue, ShellError, Value};
+
+pub fn extract_strings(value: Value) -> Result<Vec<String>, ShellError> {
+    match (
+        <String as FromValue>::from_value(&value),
+        <Vec<String> as FromValue>::from_value(&value),
+    ) {
+        (Ok(col), Err(_)) => Ok(vec![col]),
+        (Err(_), Ok(cols)) => Ok(cols),
+        _ => Err(ShellError::IncompatibleParametersSingle(
+            "Expected a string or list of strings".into(),
+            value.span()?,
+        )),
+    }
+}

crates/nu-command/src/database/mod.rs

@@ -1,3 +1,5 @@
-mod sqlite;
+mod commands;
+mod values;
 
-pub use sqlite::SQLiteDatabase;
+pub use commands::add_database_decls;
+pub(crate) use values::SQLiteDatabase;

crates/nu-command/src/database/values/mod.rs

@@ -0,0 +1,3 @@
+mod sqlite;
+
+pub(crate) use sqlite::SQLiteDatabase;

crates/nu-command/src/database/values/sqlite.rs

@@ -1,35 +1,188 @@
-use std::path::{Path, PathBuf};
+use std::{
+    fs::File,
+    io::Read,
+    path::{Path, PathBuf},
+};
 
-use nu_protocol::{CustomValue, ShellError, Span, Spanned, Value};
+use nu_protocol::{CustomValue, PipelineData, ShellError, Span, Spanned, Value};
 use rusqlite::{types::ValueRef, Connection, Row};
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
 
-#[derive(Debug, Serialize, Deserialize)]
+use sqlparser::ast::Query;
+
+const SQLITE_MAGIC_BYTES: &[u8] = "SQLite format 3\0".as_bytes();
+
+#[derive(Debug)]
 pub struct SQLiteDatabase {
     // I considered storing a SQLite connection here, but decided against it because
     // 1) YAGNI, 2) it's not obvious how cloning a connection could work, 3) state
     // management gets tricky quick. Revisit this approach if we find a compelling use case.
     path: PathBuf,
+    pub query: Option<Query>,
+}
+
+// Mocked serialization of the object
+impl Serialize for SQLiteDatabase {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_none()
+    }
+}
+
+// Mocked deserialization of the object
+impl<'de> Deserialize<'de> for SQLiteDatabase {
+    fn deserialize<D>(_deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let path = std::path::Path::new("");
+        Ok(SQLiteDatabase::new(path))
+    }
 }
 
 impl SQLiteDatabase {
-    pub fn new(path: &Path) -> SQLiteDatabase {
-        SQLiteDatabase {
+    pub fn new(path: &Path) -> Self {
+        Self {
             path: PathBuf::from(path),
+            query: None,
         }
     }
 
+    pub fn try_from_path(path: &Path, span: Span) -> Result<Self, ShellError> {
+        let mut file = File::open(path).map_err(|e| {
+            ShellError::GenericError(
+                "Error opening file".into(),
+                e.to_string(),
+                Some(span),
+                None,
+                Vec::new(),
+            )
+        })?;
+
+        let mut buf: [u8; 16] = [0; 16];
+        file.read_exact(&mut buf)
+            .map_err(|e| {
+                ShellError::GenericError(
+                    "Error reading file header".into(),
+                    e.to_string(),
+                    Some(span),
+                    None,
+                    Vec::new(),
+                )
+            })
+            .and_then(|_| {
+                if buf == SQLITE_MAGIC_BYTES {
+                    Ok(SQLiteDatabase::new(path))
+                } else {
+                    Err(ShellError::GenericError(
+                        "Error reading file".into(),
+                        "Not a SQLite file".into(),
+                        Some(span),
+                        None,
+                        Vec::new(),
+                    ))
+                }
+            })
+    }
+
     pub fn query(&self, sql: &Spanned<String>, call_span: Span) -> Result<Value, ShellError> {
         let db = open_sqlite_db(&self.path, call_span)?;
-        to_shell_error(
-            run_sql_query(db, sql),
-            "Failed to query SQLite database",
-            sql.span,
-        )
+        run_sql_query(db, sql).map_err(|e| {
+            ShellError::GenericError(
+                "Failed to query SQLite database".into(),
+                e.to_string(),
+                Some(sql.span),
+                None,
+                Vec::new(),
+            )
+        })
     }
 
-    pub fn describe(&self) -> String {
-        format!("A SQLite database at {:?}", self.path)
+    pub fn collect(&self, call_span: Span) -> Result<Value, ShellError> {
+        let sql = match &self.query {
+            Some(query) => Ok(format!("{}", query)),
+            None => Err(ShellError::GenericError(
+                "Error collecting from db".into(),
+                "No query found in connection".into(),
+                Some(call_span),
+                None,
+                Vec::new(),
+            )),
+        }?;
+
+        let sql = Spanned {
+            item: sql,
+            span: call_span,
+        };
+
+        let db = open_sqlite_db(&self.path, call_span)?;
+        run_sql_query(db, &sql).map_err(|e| {
+            ShellError::GenericError(
+                "Failed to query SQLite database".into(),
+                e.to_string(),
+                Some(sql.span),
+                None,
+                Vec::new(),
+            )
+        })
+    }
+
+    pub fn try_from_value(value: Value) -> Result<Self, ShellError> {
+        match value {
+            Value::CustomValue { val, span } => match val.as_any().downcast_ref::<Self>() {
+                Some(db) => Ok(Self {
+                    path: db.path.clone(),
+                    query: db.query.clone(),
+                }),
+                None => Err(ShellError::CantConvert(
+                    "database".into(),
+                    "non-database".into(),
+                    span,
+                    None,
+                )),
+            },
+            x => Err(ShellError::CantConvert(
+                "database".into(),
+                x.get_type().to_string(),
+                x.span()?,
+                None,
+            )),
+        }
+    }
+
+    pub fn try_from_pipeline(input: PipelineData, span: Span) -> Result<Self, ShellError> {
+        let value = input.into_value(span);
+        Self::try_from_value(value)
+    }
+
+    pub fn into_value(self, span: Span) -> Value {
+        Value::CustomValue {
+            val: Box::new(self),
+            span,
+        }
+    }
+
+    pub fn describe(&self, span: Span) -> Value {
+        let cols = vec!["connection".to_string(), "query".to_string()];
+        let connection = Value::String {
+            val: self.path.to_str().unwrap_or("").to_string(),
+            span,
+        };
+
+        let query = match &self.query {
+            Some(query) => format!("{query}"),
+            None => "".into(),
+        };
+
+        let query = Value::String { val: query, span };
+
+        Value::Record {
+            cols,
+            vals: vec![connection, query],
+            span,
+        }
     }
 }
 
@@ -37,6 +190,7 @@ impl CustomValue for SQLiteDatabase {
     fn clone_value(&self, span: Span) -> Value {
         let cloned = SQLiteDatabase {
             path: self.path.clone(),
+            query: self.query.clone(),
         };
 
         Value::CustomValue {
@@ -51,11 +205,15 @@ impl CustomValue for SQLiteDatabase {
 
     fn to_base_value(&self, span: Span) -> Result<Value, ShellError> {
         let db = open_sqlite_db(&self.path, span)?;
-        to_shell_error(
-            read_entire_sqlite_db(db, span),
-            "Failed to read from SQLite database",
-            span,
-        )
+        read_entire_sqlite_db(db, span).map_err(|e| {
+            ShellError::GenericError(
+                "Failed to read from SQLite database".into(),
+                e.to_string(),
+                Some(span),
+                None,
+                Vec::new(),
+            )
+        })
     }
 
     fn as_any(&self) -> &dyn std::any::Any {
@@ -70,11 +228,15 @@ impl CustomValue for SQLiteDatabase {
     fn follow_path_string(&self, _column_name: String, span: Span) -> Result<Value, ShellError> {
         let db = open_sqlite_db(&self.path, span)?;
 
-        to_shell_error(
-            read_single_table(db, _column_name, span),
-            "Failed to read from SQLite database",
-            span,
-        )
+        read_single_table(db, _column_name, span).map_err(|e| {
+            ShellError::GenericError(
+                "Failed to read from SQLite database".into(),
+                e.to_string(),
+                Some(span),
+                None,
+                Vec::new(),
+            )
+        })
     }
 
     fn typetag_name(&self) -> &'static str {
@@ -86,31 +248,52 @@ impl CustomValue for SQLiteDatabase {
     }
 }
 
-// TODO: is there a more elegant way to map rusqlite errors to ShellErrors?
-fn to_shell_error<T>(
-    result: Result<T, rusqlite::Error>,
-    message: &str,
-    span: Span,
-) -> Result<T, nu_protocol::ShellError> {
-    match result {
-        Ok(val) => Ok(val),
-        Err(err) => Err(ShellError::GenericError(
-            message.to_string(),
-            err.to_string(),
-            Some(span),
-            None,
-            Vec::new(),
-        )),
-    }
-}
-
 fn open_sqlite_db(path: &Path, call_span: Span) -> Result<Connection, nu_protocol::ShellError> {
     let path = path.to_string_lossy().to_string();
-    to_shell_error(
-        Connection::open(path),
-        "Failed to open SQLite database",
-        call_span,
-    )
+
+    Connection::open(path).map_err(|e| {
+        ShellError::GenericError(
+            "Failed to open SQLite database".into(),
+            e.to_string(),
+            Some(call_span),
+            None,
+            Vec::new(),
+        )
+    })
+}
+
+fn run_sql_query(conn: Connection, sql: &Spanned<String>) -> Result<Value, rusqlite::Error> {
+    let mut stmt = conn.prepare(&sql.item)?;
+    let results = stmt.query([])?;
+
+    let nu_records = results
+        .mapped(|row| Result::Ok(convert_sqlite_row_to_nu_value(row, sql.span)))
+        .into_iter()
+        .collect::<Result<Vec<Value>, rusqlite::Error>>()?;
+
+    Ok(Value::List {
+        vals: nu_records,
+        span: sql.span,
+    })
+}
+
+fn read_single_table(
+    conn: Connection,
+    table_name: String,
+    call_span: Span,
+) -> Result<Value, rusqlite::Error> {
+    let mut stmt = conn.prepare(&format!("SELECT * FROM {}", table_name))?;
+    let results = stmt.query([])?;
+
+    let nu_records = results
+        .mapped(|row| Result::Ok(convert_sqlite_row_to_nu_value(row, call_span)))
+        .into_iter()
+        .collect::<Result<Vec<Value>, rusqlite::Error>>()?;
+
+    Ok(Value::List {
+        vals: nu_records,
+        span: call_span,
+    })
 }
 
 fn read_entire_sqlite_db(conn: Connection, call_span: Span) -> Result<Value, rusqlite::Error> {
@@ -146,41 +329,6 @@ fn read_entire_sqlite_db(conn: Connection, call_span: Span) -> Result<Value, rus
         span: call_span,
     })
 }
-
-fn run_sql_query(conn: Connection, sql: &Spanned<String>) -> Result<Value, rusqlite::Error> {
-    let mut stmt = conn.prepare(&sql.item)?;
-    let mut results = stmt.query([])?;
-    let mut nu_records = Vec::new();
-
-    while let Some(table_row) = results.next()? {
-        nu_records.push(convert_sqlite_row_to_nu_value(table_row, sql.span))
-    }
-
-    Ok(Value::List {
-        vals: nu_records,
-        span: sql.span,
-    })
-}
-
-fn read_single_table(
-    conn: Connection,
-    table_name: String,
-    call_span: Span,
-) -> Result<Value, rusqlite::Error> {
-    let mut stmt = conn.prepare(&format!("SELECT * FROM {}", table_name))?;
-    let mut results = stmt.query([])?;
-    let mut nu_records = Vec::new();
-
-    while let Some(table_row) = results.next()? {
-        nu_records.push(convert_sqlite_row_to_nu_value(table_row, call_span))
-    }
-
-    Ok(Value::List {
-        vals: nu_records,
-        span: call_span,
-    })
-}
-
 fn convert_sqlite_row_to_nu_value(row: &Row, span: Span) -> Value {
     let mut vals = Vec::new();
     let colnamestr = row.as_ref().column_names().to_vec();