Merge 'Ensure that Connection::query() checks whether its schema is up to date' from Jussi Saurio

Closes #2997
Fixes issue #2997 where connection 2 cannot see tables created by
another connection 1, because `Connection::query()` was not checking
whether its copy of the schema was stale.

Reviewed-by: Preston Thorpe <preston@turso.tech>

Closes #3008

core/lib.rs

@@ -1106,6 +1106,7 @@ impl Connection {
                 "The supplied SQL string contains no statements".to_string(),
             ));
         }
+        self.maybe_update_schema()?;
         let sql = sql.as_ref();
         tracing::trace!("Preparing and executing batch: {}", sql);
         let mut parser = Parser::new(sql.as_bytes());
@@ -1143,6 +1144,7 @@ impl Connection {
             return Err(LimboError::InternalError("Connection closed".to_string()));
         }
         let sql = sql.as_ref();
+        self.maybe_update_schema()?;
         tracing::trace!("Querying: {}", sql);
         let mut parser = Parser::new(sql.as_bytes());
         let cmd = parser.next_cmd()?;
@@ -1216,6 +1218,7 @@ impl Connection {
             return Err(LimboError::InternalError("Connection closed".to_string()));
         }
         let sql = sql.as_ref();
+        self.maybe_update_schema()?;
         let mut parser = Parser::new(sql.as_bytes());
         while let Some(cmd) = parser.next_cmd()? {
             let syms = self.syms.borrow();
@@ -1224,7 +1227,6 @@ impl Connection {
             let input = str::from_utf8(&sql.as_bytes()[..byte_offset_end])
                 .unwrap()
                 .trim();
-            self.maybe_update_schema()?;
             match cmd {
                 Cmd::Explain(stmt) => {
                     let program = translate::translate(

tests/integration/query_processing/test_transactions.rs

@@ -123,3 +123,52 @@ fn test_txn_error_doesnt_rollback_txn() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+/// Connection 2 should see the initial data (table 'test' in schema + 2 rows). Regression test for #2997
+/// It should then see another created table 'test2' in schema, as well.
+fn test_transaction_visibility() {
+    let tmp_db = TempDatabase::new("test_transaction_visibility.db", true);
+    let conn1 = tmp_db.connect_limbo();
+    let conn2 = tmp_db.connect_limbo();
+
+    conn1
+        .execute("CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)")
+        .unwrap();
+
+    conn1
+        .execute("INSERT INTO test (id, value) VALUES (1, 'initial')")
+        .unwrap();
+
+    let mut stmt = conn2.query("SELECT COUNT(*) FROM test").unwrap().unwrap();
+    loop {
+        match stmt.step().unwrap() {
+            StepResult::Row => {
+                let row = stmt.row().unwrap();
+                assert_eq!(*row.get::<&Value>(0).unwrap(), Value::Integer(1));
+            }
+            StepResult::IO => stmt.run_once().unwrap(),
+            StepResult::Done => break,
+            StepResult::Busy => panic!("database is busy"),
+            StepResult::Interrupt => panic!("interrupted"),
+        }
+    }
+
+    conn1
+        .execute("CREATE TABLE test2 (id INTEGER PRIMARY KEY, value TEXT)")
+        .unwrap();
+
+    let mut stmt = conn2.query("SELECT COUNT(*) FROM test2").unwrap().unwrap();
+    loop {
+        match stmt.step().unwrap() {
+            StepResult::Row => {
+                let row = stmt.row().unwrap();
+                assert_eq!(*row.get::<&Value>(0).unwrap(), Value::Integer(0));
+            }
+            StepResult::IO => stmt.run_once().unwrap(),
+            StepResult::Done => break,
+            StepResult::Busy => panic!("database is busy"),
+            StepResult::Interrupt => panic!("interrupted"),
+        }
+    }
+}