Merge 'Fix MVCC startup infinite loop when using existing DB' from Jussi Saurio

MVCC bootstrap connection got stuck into an infinite statement reparsing
loop because the bootstrap procedure happened before the on-disk schema
was deserialized.
closes #3518

Closes #3522

core/lib.rs

@@ -451,15 +451,6 @@ impl Database {
             n_connections: AtomicUsize::new(0),
         });
 
-        if opts.enable_mvcc {
-            let mv_store = db.mv_store.as_ref().unwrap();
-            let mvcc_bootstrap_conn = db.connect_mvcc_bootstrap()?;
-            mv_store.bootstrap(mvcc_bootstrap_conn)?;
-        }
-
-        db.register_global_builtin_extensions()
-            .expect("unable to register global extensions");
-
         // Check: https://github.com/tursodatabase/turso/pull/1761#discussion_r2154013123
         if db_state.is_initialized() {
             // parse schema
@@ -502,6 +493,15 @@ impl Database {
             })?;
         }
 
+        if opts.enable_mvcc {
+            let mv_store = db.mv_store.as_ref().unwrap();
+            let mvcc_bootstrap_conn = db.connect_mvcc_bootstrap()?;
+            mv_store.bootstrap(mvcc_bootstrap_conn)?;
+        }
+
+        db.register_global_builtin_extensions()
+            .expect("unable to register global extensions");
+
         Ok(db)
     }
 

tests/integration/query_processing/test_transactions.rs

@@ -646,6 +646,43 @@ fn test_mvcc_recovery_of_both_checkpointed_and_noncheckpointed_tables_works() {
     assert_eq!(rows, expected2);
 }
 
+#[test]
+fn test_non_mvcc_to_mvcc() {
+    // Create non-mvcc database
+    let tmp_db = TempDatabase::new("test_non_mvcc_to_mvcc.db", false);
+    let conn = tmp_db.connect_limbo();
+
+    // Create table and insert data
+    execute_and_log(
+        &conn,
+        "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)",
+    )
+    .unwrap();
+    execute_and_log(&conn, "INSERT INTO test VALUES (1, 'hello')").unwrap();
+
+    // Checkpoint to persist changes
+    execute_and_log(&conn, "PRAGMA wal_checkpoint(TRUNCATE)").unwrap();
+
+    let path = tmp_db.path.clone();
+    drop(conn);
+    drop(tmp_db);
+
+    // Reopen in mvcc mode
+    let tmp_db = TempDatabase::new_with_existent_with_opts(
+        &path,
+        turso_core::DatabaseOpts::new().with_mvcc(true),
+    );
+    let conn = tmp_db.connect_limbo();
+
+    // Query should work
+    let stmt = query_and_log(&conn, "SELECT * FROM test").unwrap().unwrap();
+    let rows = helper_read_all_rows(stmt);
+
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0][0], Value::Integer(1));
+    assert_eq!(rows[0][1], Value::Text("hello".into()));
+}
+
 fn helper_read_all_rows(mut stmt: turso_core::Statement) -> Vec<Vec<Value>> {
     let mut ret = Vec::new();
     loop {