refactor

tools/iface/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "conduit_iface"
+version = "0.1.0"
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+itertools = "0.10.1"
+sled = { version = "0.34.6", features = ["compression", "no_metrics"] }
+rusqlite = { version = "0.25.3", features = ["bundled"] }
+anyhow = "1.0.42"

tools/iface/src/db.rs

@@ -0,0 +1,43 @@
+pub mod sled;
+pub mod sqlite;
+
+use itertools::Itertools;
+
+pub type KVIter<'a> = Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a>;
+
+pub type TreeKVIter<'a> = Box<dyn Iterator<Item = (Vec<u8>, KVIter<'a>)> + 'a>;
+
+pub trait Database {
+    fn iter<'a>(&'a self) -> TreeKVIter<'a>;
+
+    fn segment<'a>(&'a mut self, name: Vec<u8>) -> Option<Box<dyn Segment + 'a>>; // change return type to Result
+}
+
+pub trait Segment {
+    fn batch_insert<'a>(
+        &'a mut self,
+        batch: Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a>,
+    ) -> anyhow::Result<()>;
+}
+
+pub fn copy_database(
+    src: &impl Database,
+    dst: &mut impl Database,
+    chunk_size: usize,
+) -> anyhow::Result<()> {
+    for (tree, i) in src.iter() {
+        dbg!(&tree);
+
+        let mut t = dst.segment(tree).unwrap(); // todo remove unwrap
+
+        let mut x: usize = 0;
+
+        for chunk in &i.chunks(chunk_size) {
+            dbg!(&x);
+            t.batch_insert(Box::new(chunk))?;
+            x += chunk_size;
+        }
+    }
+
+    Ok(())
+}

tools/iface/src/db/sled.rs

@@ -0,0 +1,68 @@
+use std::path::Path;
+
+use sled::{Batch, Config, Db, Result, Tree};
+
+use super::{Database, KVIter, Segment, TreeKVIter};
+
+pub fn new_db<P: AsRef<Path>>(path: P) -> Result<Db> {
+    Config::default().path(path).use_compression(true).open()
+}
+
+pub struct SledDB(Db);
+
+impl SledDB {
+    pub fn new(db: Db) -> Self {
+        Self(db)
+    }
+}
+
+impl Database for SledDB {
+    fn iter<'a>(&'a self) -> TreeKVIter<'a> {
+        Box::new(
+            self.0
+                .tree_names()
+                .into_iter()
+                .map(|v| v.to_vec())
+                .filter_map(move |v| {
+                    if let Ok(t) = self.0.open_tree(&v) {
+                        Some((v, t))
+                    } else {
+                        None
+                    }
+                })
+                .map(|(v, t): (Vec<u8>, Tree)| -> (Vec<u8>, KVIter<'a>) {
+                    let i = t.into_iter().filter_map(|r| {
+                        if let Ok(t) = r {
+                            Some((t.0.to_vec(), t.1.to_vec()))
+                        } else {
+                            None
+                        }
+                    });
+
+                    (v, Box::new(i))
+                }),
+        )
+    }
+
+    fn segment(&mut self, name: Vec<u8>) -> Option<Box<dyn Segment>> {
+        self.0
+            .open_tree(name)
+            .ok()
+            .map(|t| -> Box<dyn Segment> { Box::new(t) })
+    }
+}
+
+impl Segment for Tree {
+    fn batch_insert(
+        &mut self,
+        batch: Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + '_>,
+    ) -> anyhow::Result<()> {
+        let mut sled_batch = Batch::default();
+
+        for (k, v) in batch {
+            sled_batch.insert(k, v)
+        }
+
+        self.apply_batch(sled_batch).map_err(Into::into)
+    }
+}

tools/iface/src/db/sqlite.rs

@@ -2,6 +2,8 @@ use std::path::Path;
 
 use rusqlite::{self, Connection, DatabaseName::Main};
 
+use super::{Database, Segment};
+
 pub fn new_conn<P: AsRef<Path>>(path: P) -> rusqlite::Result<Connection> {
     let path = path.as_ref().join("conduit.db");
     let conn = Connection::open(path)?;
@@ -17,22 +19,29 @@ impl SqliteDB {
     pub fn new(conn: Connection) -> Self {
         Self(conn)
     }
+}
 
-    pub fn table<'a>(&'a mut self, string: String) -> rusqlite::Result<SqliteTable<'a>> {
+impl Database for SqliteDB {
+    fn iter<'a>(&'a self) -> super::TreeKVIter<'a> {
+        todo!("iterate over tables, pick only tables that have columns 'key' and 'value', then iterate over that with values")
+    }
+
+    fn segment<'a>(&'a mut self, name: Vec<u8>) -> Option<Box<dyn Segment + 'a>> {
+        let string = String::from_utf8(name).unwrap();
         // taken from src/database/abstraction/sqlite.rs
-        self.0.execute(format!("CREATE TABLE IF NOT EXISTS {} ( \"key\" BLOB PRIMARY KEY, \"value\" BLOB NOT NULL )", &string).as_str(), [])?;
+        self.0.execute(format!("CREATE TABLE IF NOT EXISTS {} ( \"key\" BLOB PRIMARY KEY, \"value\" BLOB NOT NULL )", &string).as_str(), []).unwrap();
 
-        Ok(SqliteTable(&mut self.0, string))
+        Some(Box::new(SqliteTable(&mut self.0, string)))
     }
 }
 
 pub struct SqliteTable<'a>(&'a mut Connection, String);
 
-impl SqliteTable<'_> {
-    pub fn batch_insert(
+impl Segment for SqliteTable<'_> {
+    fn batch_insert(
         &mut self,
-        batch: impl Iterator<Item = (Vec<u8>, Vec<u8>)>,
-    ) -> rusqlite::Result<()> {
+        batch: Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + '_>,
+    ) -> anyhow::Result<()> {
         let tx = self.0.transaction()?;
         let sql_s = format!("INSERT INTO {} (key, value) VALUES (?, ?)", &self.1);
         let sql = sql_s.as_str();
@@ -41,6 +50,6 @@ impl SqliteTable<'_> {
             tx.execute(sql, rusqlite::params![k, v])?;
         }
 
-        tx.commit()
+        tx.commit().map_err(Into::into)
     }
 }

tools/iface/src/lib.rs

@@ -0,0 +1 @@
+pub mod db;

tools/sled_to_sqlite/Cargo.toml

@@ -6,8 +6,6 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-sled = { version = "0.34.6", features = ["compression", "no_metrics"] }
-rusqlite = { version = "0.25.3", features = ["bundled"] }
 clap = "2.33.3"
 anyhow = "1.0.41"
-itertools = "0.10.1"
+conduit_iface = { path = "../iface/" }

tools/sled_to_sqlite/src/main.rs

@@ -1,12 +1,8 @@
-mod sled;
-mod sqlite;
 use std::path::Path;
 
 use clap::{App, Arg};
 
-use crate::{sled::SledDB, sqlite::SqliteDB};
-
-use itertools::Itertools;
+use conduit_iface::db::{copy_database, sled, sqlite};
 
 fn main() -> anyhow::Result<()> {
     let matches = App::new("Conduit Sled to Sqlite Migrator")
@@ -46,25 +42,11 @@ fn main() -> anyhow::Result<()> {
 
     dbg!(&source_dir, &dest_dir);
 
-    let sled = SledDB::new(crate::sled::new_db(source_dir)?);
+    let sled = sled::SledDB::new(sled::new_db(source_dir)?);
 
-    let mut sqlite = SqliteDB::new(sqlite::new_conn(dest_dir)?);
+    let mut sqlite = sqlite::SqliteDB::new(sqlite::new_conn(dest_dir)?);
 
-    for (tree, i) in sled.iter() {
-        let tree = String::from_utf8(tree)?;
-
-        dbg!(&tree);
-
-        let mut t = sqlite.table(tree)?;
-
-        let mut x: u32 = 0;
-
-        for chunk in &i.chunks(1000) {
-            dbg!(&x);
-            t.batch_insert(chunk)?;
-            x += 1000;
-        }
-    }
+    copy_database(&sled, &mut sqlite, 1000)?;
 
     Ok(())
 }

tools/sled_to_sqlite/src/sled.rs

@@ -1,44 +0,0 @@
-use std::path::Path;
-
-use sled;
-
-pub fn new_db<P: AsRef<Path>>(path: P) -> sled::Result<sled::Db> {
-    sled::Config::default()
-        .path(path)
-        .use_compression(true)
-        .open()
-}
-
-pub struct SledDB(sled::Db);
-
-impl SledDB {
-    pub fn iter<'a>(
-        &'a self,
-    ) -> impl Iterator<Item = (Vec<u8>, impl Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a)> {
-        self.0
-            .tree_names()
-            .into_iter()
-            .map(|v| v.to_vec())
-            .filter_map(move |v| {
-                let t = if let Ok(t) = self.0.open_tree(&v) {
-                    t
-                } else {
-                    return None;
-                };
-
-                let i = t.into_iter().filter_map(|r| {
-                    if let Ok(t) = r {
-                        Some((t.0.to_vec(), t.1.to_vec()))
-                    } else {
-                        None
-                    }
-                });
-
-                Some((v, i))
-            })
-    }
-
-    pub fn new(db: sled::Db) -> Self {
-        Self(db)
-    }
-}