WIP renumber partition when performing a fast-forward

Author: Grisha Kruglov

db/src/lib.rs

@@ -71,6 +71,8 @@ pub use bootstrap::{
     CORE_SCHEMA_VERSION,
 };
 
+pub use renumber::renumber;
+
 use edn::symbols;
 
 pub use entids::{
@@ -106,6 +108,7 @@ pub use tx_observer::{
 pub use types::{
     AttributeSet,
     DB,
+    Partition,
     PartitionMap,
     TransactableValue,
     TxReport,

src/sync.rs

@@ -22,16 +22,11 @@ use errors::{
 };
 
 use mentat_core::{
-    Entid,
     KnownEntid,
 };
 use mentat_db::{
-    USER0,
-    TX0,
-};
-use mentat_db::db;
-use mentat_db::db::{
-    PartitionMapping,
+    renumber,
+    PartitionMap,
 };
 
 use entity_builder::{
@@ -52,11 +47,7 @@ pub trait Syncable {
     fn sync(&mut self, server_uri: &String, user_uuid: &String) -> Result<()>;
 }
 
-fn within_user_partition(entid: Entid) -> bool {
-    entid >= USER0 && entid < TX0
-}
-
-fn fast_forward_local<'a, 'c>(in_progress: &mut InProgress<'a, 'c>, txs: Vec<Tx>) -> Result<()> {
+fn fast_forward_local<'a, 'c>(in_progress: &mut InProgress<'a, 'c>, txs: Vec<Tx>) -> Result<Option<PartitionMap>> {
     let mut last_tx = None;
 
     // During fast-forwarding, we will insert datoms with known entids
@@ -69,21 +60,22 @@ fn fast_forward_local<'a, 'c>(in_progress: &mut InProgress<'a, 'c>, txs: Vec<Tx>
     // In absence of excision and implementation bugs, this should work
     // just as if we counted number of incoming entids and expanded by
     // that number instead.
-    let mut largest_entid_encountered = USER0;
+    let mut last_encountered_partition_map = None;
 
     for tx in txs {
         let mut builder = TermBuilder::new();
 
+        last_encountered_partition_map = match tx.parts[0].partitions.clone() {
+            Some(parts) => Some(parts),
+            None => bail!(TolstoyError::BadServerState("Missing partition map in incoming transaction".to_string()))
+        };
+
         for part in tx.parts {
             if part.added {
                 builder.add(KnownEntid(part.e), KnownEntid(part.a), part.v)?;
             } else {
                 builder.retract(KnownEntid(part.e), KnownEntid(part.a), part.v)?;
             }
-            // Ignore datoms that fall outside of the user partition:
-            if within_user_partition(part.e) && part.e > largest_entid_encountered {
-                largest_entid_encountered = part.e;
-            }
         }
 
         let report = in_progress.transact_builder(builder)?;
@@ -96,15 +88,9 @@ fn fast_forward_local<'a, 'c>(in_progress: &mut InProgress<'a, 'c>, txs: Vec<Tx>
     if let Some((entid, uuid)) = last_tx {
         SyncMetadataClient::set_remote_head(&mut in_progress.transaction, &uuid)?;
         TxMapper::set_tx_uuid(&mut in_progress.transaction, entid, &uuid)?;
-
-        // We only need to advance the user partition, since we're using KnownEntid and partition
-        // won't get auto-updated; shouldn't be a problem for tx partition, since we're relying on
-        // the builder to create a tx and advance the partition for us.
-        in_progress.partition_map.expand_up_to(":db.part/user", largest_entid_encountered);
-        db::update_partition_map(&mut in_progress.transaction, &in_progress.partition_map)?;
     }
 
-    Ok(())
+    Ok(last_encountered_partition_map)
 }
 
 impl Conn {
@@ -118,6 +104,7 @@ impl Conn {
         let mut in_progress = self.begin_transaction(sqlite)?;
 
         let sync_result = Syncer::flow(&mut in_progress.transaction, server_uri, &uuid)?;
+        let mut incoming_partition = None;
 
         match sync_result {
             SyncResult::EmptyServer => (),
@@ -126,7 +113,10 @@ impl Conn {
             SyncResult::Merge => bail!(TolstoyError::NotYetImplemented(
                 format!("Can't sync against diverged local.")
             )),
-            SyncResult::LocalFastForward(txs) => fast_forward_local(&mut in_progress, txs)?,
+            SyncResult::LocalFastForward(txs) => {
+                incoming_partition = fast_forward_local(&mut in_progress, txs)?;
+                ()
+            },
             SyncResult::BadServerState => bail!(TolstoyError::NotYetImplemented(
                 format!("Bad server state.")
             )),
@@ -136,6 +126,16 @@ impl Conn {
             )),
         }
 
+        match incoming_partition {
+            Some(incoming) => {
+                let root = SyncMetadataClient::get_partitions(&in_progress.transaction, true)?;
+                let current = SyncMetadataClient::get_partitions(&in_progress.transaction, false)?;
+                renumber(&in_progress.transaction, &root, &current, &incoming)?;
+                ()
+            },
+            None => ()
+        }
+
         in_progress.commit()
     }
 }

tolstoy/src/metadata.rs

@@ -47,8 +47,12 @@ impl SyncMetadataClient {
         Ok(())
     }
 
-    pub fn get_partitions(tx: &rusqlite::Transaction) -> Result<PartitionMap> {
-        let mut stmt: ::rusqlite::Statement = tx.prepare("SELECT part, start, idx FROM tolstoy_parts")?;
+    pub fn get_partitions(tx: &rusqlite::Transaction, core: bool) -> Result<PartitionMap> {
+        let db_table = match core {
+            true => "parts",
+            false => "tolstoy_parts"
+        };
+        let mut stmt: ::rusqlite::Statement = tx.prepare(&format!("SELECT part, start, idx FROM {}", db_table))?;
         let m: Result<PartitionMap> = stmt.query_and_then(&[], |row| -> Result<(String, Partition)> {
             Ok((row.get_checked(0)?, Partition::new(row.get_checked(1)?, row.get_checked(2)?)))
         })?.collect();

tolstoy/src/schema.rs

@@ -97,7 +97,7 @@ pub mod tests {
             (_, _) => { panic!("Wrong number of results."); },
         }
 
-        let partitions = SyncMetadataClient::get_partitions(&tx).unwrap();
+        let partitions = SyncMetadataClient::get_partitions(&tx, false).unwrap();
 
         assert_eq!(partitions.len(), 1);
 
@@ -146,7 +146,7 @@ pub mod tests {
             (_, _) => { panic!("Wrong number of results."); },
         }
 
-        let partitions = SyncMetadataClient::get_partitions(&tx).unwrap();
+        let partitions = SyncMetadataClient::get_partitions(&tx, false).unwrap();
 
         assert_eq!(partitions.len(), 1);
 

tolstoy/src/syncer.rs

@@ -19,6 +19,8 @@ use mentat_core::{
 use mentat_db::{
     CORE_SCHEMA_VERSION,
     USER0,
+    Partition,
+    PartitionMap,
 };
 use bootstrap::{
     BootstrapHelper,
@@ -45,8 +47,6 @@ use tx_mapper::{
     TxMapper,
 };
 use types::{
-    Partition,
-    PartitionMap,
     Tx,
     TxPart,
 };
@@ -164,7 +164,7 @@ impl<'c> TxReceiver for UploadingTxReceiver<'c> {
         // so this is temporary and for development purposes only.
         let mut tx_partition_map = PartitionMap::default();
         tx_partition_map.insert(PARTITION_USER.to_string(), Partition::new(USER0, largest_e + 1));
-        datoms[0].parts = Some(tx_partition_map);
+        datoms[0].partitions = Some(tx_partition_map);
 
         // Upload all chunks.
         for datom in &datoms {

tolstoy/src/tx_processor.rs

@@ -95,7 +95,7 @@ where T: Sized + Iterator<Item=Result<TxPart>> + 't {
                 Err(_) => None,
                 Ok(datom) => {
                     Some(TxPart {
-                        parts: None,
+                        partitions: None,
                         e: datom.e,
                         a: datom.a,
                         v: datom.v.clone(),
@@ -113,7 +113,7 @@ where T: Sized + Iterator<Item=Result<TxPart>> + 't {
 
 fn to_tx_part(row: &rusqlite::Row) -> Result<TxPart> {
     Ok(TxPart {
-        parts: None,
+        partitions: None,
         e: row.get_checked(0)?,
         a: row.get_checked(1)?,
         v: TypedValue::from_sql_value_pair(row.get_checked(2)?, row.get_checked(3)?)?,

tolstoy/src/types.rs

@@ -15,6 +15,8 @@ use mentat_core::{
     TypedValue,
 };
 
+use mentat_db::PartitionMap;
+
 // For returning out of the downloader as an ordered list.
 #[derive(Debug, Clone)]
 pub struct Tx {
@@ -26,7 +28,7 @@ pub struct Tx {
 pub struct TxPart {
     // TODO this is a temporary for development. Only first TxPart in a chunk series should have a non-None 'parts'.
     // 'parts' should actually live in a transaction, but we do this now to avoid changing the server until dust settles.
-    pub parts: Option<PartitionMap>,
+    pub partitions: Option<PartitionMap>,
     pub e: Entid,
     pub a: Entid,
     pub v: TypedValue,