mvcc: add MvStore::get_next_table_id()

this assigns "synthetic" root page numbers instead of using the pager
or btree cursors to allocate pages. Actual page allocation will only
happen during checkpoint, since regular MVCC commit bypasses the
pager.

core/mvcc/database/mod.rs

@@ -793,6 +793,7 @@ pub struct MvStore<Clock: LogicalClock> {
     txs: SkipMap<TxID, Transaction>,
     tx_ids: AtomicU64,
     next_rowid: AtomicU64,
+    next_table_id: AtomicU64,
     clock: Clock,
     storage: Storage,
     loaded_tables: RwLock<HashSet<u64>>,
@@ -823,6 +824,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
             txs: SkipMap::new(),
             tx_ids: AtomicU64::new(1), // let's reserve transaction 0 for special purposes
             next_rowid: AtomicU64::new(0), // TODO: determine this from B-Tree
+            next_table_id: AtomicU64::new(2), // table id 1 / root page 1 is always sqlite_schema.
             clock,
             storage,
             loaded_tables: RwLock::new(HashSet::new()),
@@ -835,6 +837,14 @@ impl<Clock: LogicalClock> MvStore<Clock> {
         }
     }
 
+    /// MVCC does not use the pager/btree cursors to create pages until checkpoint.
+    /// This method is used to assign root page numbers when Insn::CreateBtree is used.
+    /// NOTE: during MVCC recovery (not implemented yet), [MvStore::next_table_id] must be
+    /// initialized to the current highest table id / root page number.
+    pub fn get_next_table_id(&self) -> u64 {
+        self.next_table_id.fetch_add(1, Ordering::SeqCst)
+    }
+
     pub fn get_next_rowid(&self) -> i64 {
         self.next_rowid.fetch_add(1, Ordering::SeqCst) as i64
     }
@@ -1519,7 +1529,7 @@ impl<Clock: LogicalClock> MvStore<Clock> {
         // Then, scan the disk B-tree to find existing rows
         self.scan_load_table(table_id, pager)?;
 
-        self.loaded_tables.write().insert(table_id);
+        self.mark_table_as_loaded(table_id);
 
         Ok(())
     }

core/vdbe/execute.rs

@@ -6629,6 +6629,13 @@ pub fn op_create_btree(
         // TODO: implement temp databases
         todo!("temp databases not implemented yet");
     }
+
+    if let Some(mv_store) = mv_store {
+        let root_page = mv_store.get_next_table_id();
+        state.registers[*root] = Register::Value(Value::Integer(root_page as i64));
+        state.pc += 1;
+        return Ok(InsnFunctionStepResult::Step);
+    }
     // FIXME: handle page cache is full
     let root_page = return_if_io!(pager.btree_create(flags));
     state.registers[*root] = Register::Value(Value::Integer(root_page as i64));