Enable SURREALDB_CHANGE_FEED_LIVE_QUERIES for tests

core/src/cf/mutations.rs

@@ -91,17 +91,18 @@ impl TableMutation {
  h
  }
  TableMutation::SetWithDiff(_thing, current, operations) => {
- h.insert("current".to_string(), current);
- h.insert(
-  "update".to_string(),
- Value::Array(Array(
+ // derive the original
+ let mut original = current.clone();
+ original
+ .patch(Value::Array(Array(
  operations
  .clone()
  .into_iter()
- .map(|x| Value::Object(Object::from(x)))
+ .map(|op| Value::Object(Object::from(op)))
  .collect(),
- )),
- );
+ )))
+ .unwrap();
+ h.insert("update".to_string(), current);
  h
  }
  TableMutation::Del(t) => {

core/src/cf/writer.rs

@@ -2,6 +2,7 @@ use std::borrow::Cow;
 use std::collections::HashMap;
 
 use crate::cf::{TableMutation, TableMutations};
+use crate::fflags::FFLAGS;
 use crate::kvs::Key;
 use crate::sql::statements::DefineTableStatement;
 use crate::sql::thing::Thing;
@@ -75,14 +76,18 @@ impl Writer {
  current: Cow<'_, Value>,
  store_difference: bool,
  ) {
+ warn!(
+ "Recording difference for cf: id={}, previous={:?}, current={:?}, store_difference={}",
+ id, previous, current, store_difference
+ );
  if current.is_some() {
  self.buf.push(
  ns.to_string(),
  db.to_string(),
  tb.to_string(),
  match store_difference {
  true => {
- if previous.is_none() {
+ if (!FFLAGS.change_feed_live_queries.enabled()) || previous.is_none() {
  TableMutation::Set(id, current.into_owned())
  } else {
  // We intentionally record the patches in reverse (current -> previous)

core/src/fflags.rs

@@ -10,8 +10,8 @@ pub static FFLAGS: FFlags = FFlags {
  // TODO(fflag-lqcf): This TODO signature marks tests that are affected by the fflag that do not have access to the fflag (scope)
  change_feed_live_queries: FFlagEnabledStatus {
  enabled_release: false,
- enabled_debug: false,
- enabled_test: false,
+ enabled_debug: true,
+ enabled_test: true,
  env_override: "SURREALDB_CHANGE_FEED_LIVE_QUERIES",
  owner: "Hugh Kaznowski",
  description: "Disables live queries as a separate feature and moves to using change feeds as the underlying mechanism",

core/src/kvs/lq_cf.rs

@@ -159,6 +159,16 @@ impl LiveQueryTracker {
  &self,
  selector: &LqSelector,
  ) -> Vec<(LqIndexKey, LqIndexValue)> {
+ #[cfg(debug_assertions)]
+ trace!(
+ "Finding live queries for selector {:?},\n{}",
+ selector,
+ self.local_live_queries
+ .iter()
+ .map(|(k, v)| format!("{:?} -> {:?}", k, v))
+ .collect::<Vec<String>>()
+ .join("\n\n")
+ );
  Self::live_queries_for_selector_impl(&self.local_live_queries, selector)
  }
 

core/src/kvs/lq_v2_doc.rs

@@ -15,6 +15,7 @@ pub(in crate::kvs) fn construct_document(
 ) -> Result<Option<Document>, Error> {
  match mutation {
  TableMutation::Set(id, current_value) => {
+ trace!("Reconstructing document from Set {} with current value {}", id, current_value);
  let doc = Document::new_artificial(
  None,
  Some(id),
@@ -42,6 +43,7 @@ pub(in crate::kvs) fn construct_document(
  }
  TableMutation::Def(_) => Ok(None),
  TableMutation::SetWithDiff(id, current_value, operations) => {
+ trace!("Reconstructing document from SetWithDiff {} with current value {} and operations {:?}", id, current_value, operations);
  // We need a previous value otherwise the Value::compute function won't work correctly
  // This is also how IDs are carried into notifications, not via doc.rid
  let mut copy = current_value.clone();

core/src/kvs/lq_v2_fut.rs

@@ -1,3 +1,4 @@
+//! Handles the main-loop of the live query notifications on top of change feeds
 use crate::cf;
 use crate::cf::{ChangeSet, TableMutation};
 use crate::dbs::{Options, Statement};
@@ -141,7 +142,7 @@ async fn process_change_set_for_notifications(
  lq_pairs: &[(LqIndexKey, LqIndexValue)],
 ) -> Result<(), Error> {
  #[cfg(debug_assertions)]
- trace!("Moving to next change set, {:?}", change_set);
+ trace!("Moving to next change set, {:?} for {} lq_pairs", change_set, lq_pairs.len());
  for (lq_key, lq_value) in lq_pairs.iter() {
  #[cfg(debug_assertions)]
  trace!("Processing live query for notification key={:?} and value={:?}", lq_key, lq_value);

lib/tests/api/mod.rs

@@ -1074,13 +1074,15 @@ async fn changefeed() {
  let (permit, db) = new_db().await;
  db.use_ns(NS).use_db(Ulid::new().to_string()).await.unwrap();
  // Enable change feeds
- let sql = "
- DEFINE TABLE user CHANGEFEED 1h;
- ";
+ let sql = match FFLAGS.change_feed_live_queries.enabled() {
+ true => "DEFINE TABLE user CHANGEFEED 1h INCLUDE ORIGINAL;",
+ false => "DEFINE TABLE user CHANGEFEED 1h;",
+ };
  let response = db.query(sql).await.unwrap();
  drop(permit);
  response.check().unwrap();
- // Create and update users
+
+ // Create and update users - this all happens in a single changefeed entry
  let sql = "
  CREATE user:amos SET name = 'Amos';
  CREATE user:jane SET name = 'Jane';
@@ -1109,6 +1111,8 @@ async fn changefeed() {
  assert_eq!(users, expected);
  let users: Vec<RecordBuf> = db.select(table).await.unwrap();
  assert_eq!(users, expected);
+
+ // Fetch all changes
  let sql = "
  SHOW CHANGES FOR TABLE user SINCE 0 LIMIT 10;
  ";
@@ -1140,50 +1144,51 @@ async fn changefeed() {
  )
  .unwrap()
  );
- // UPDATE user:amos
+
+ // Validate CREATE user:amos
  let a = array.get(1).unwrap();
  let Value::Object(a) = a else {
  unreachable!()
  };
  let Value::Number(versionstamp1) = a.get("versionstamp").unwrap() else {
  unreachable!()
  };
+
  let changes = a.get("changes").unwrap().to_owned();
- match FFLAGS.change_feed_live_queries.enabled() {
-  true => {
-  assert_eq!(
-  changes,
-  surrealdb::sql::value(
-  r#"[
+ // The first amos is created
+ if FFLAGS.change_feed_live_queries.enabled() {
+ assert_eq!(
+ changes,
+ surrealdb::sql::value(
+ r#"[
  {
  create: {
  id: user:amos,
  name: 'Amos'
  }
  }
  ]"#
- )
- .unwrap()
- );
- }
- false => {
- assert_eq!(
- changes,
- surrealdb::sql::value(
- r#"[
+ )
+ .unwrap()
+ );
+ } else {
+ assert_eq!(
+ changes,
+ surrealdb::sql::value(
+ r#"[
  {
  update: {
  id: user:amos,
  name: 'Amos'
  }
  }
  ]"#
- )
- .unwrap()
- );
- }
+ )
+ .unwrap()
+ );
  }
- // UPDATE user:jane
+
+ // Validate CREATE user:jane
  let a = array.get(2).unwrap();
  let Value::Object(a) = a else {
  unreachable!()
@@ -1193,41 +1198,39 @@ async fn changefeed() {
  };
  assert!(versionstamp1 < versionstamp2);
  let changes = a.get("changes").unwrap().to_owned();
- match FFLAGS.change_feed_live_queries.enabled() {
- true => {
- assert_eq!(
- changes,
- surrealdb::sql::value(
- "[
+ if FFLAGS.change_feed_live_queries.enabled() {
+ assert_eq!(
+ changes,
+ surrealdb::sql::value(
+ "[
  {
  create: {
  id: user:jane,
  name: 'Jane'
  }
  }
  ]"
- )
- .unwrap()
- );
- }
- false => {
- assert_eq!(
- changes,
- surrealdb::sql::value(
- "[
+ )
+ .unwrap()
+ );
+ } else {
+ assert_eq!(
+ changes,
+ surrealdb::sql::value(
+ "[
  {
  update: {
  id: user:jane,
  name: 'Jane'
  }
  }
  ]"
- )
- .unwrap()
- );
- }
+ )
+ .unwrap()
+ );
  }
- // UPDATE user:amos
+
+ // Validate UPDATE user:amos
  let a = array.get(3).unwrap();
  let Value::Object(a) = a else {
  unreachable!()
@@ -1237,40 +1240,21 @@ async fn changefeed() {
  };
  assert!(versionstamp2 < versionstamp3);
  let changes = a.get("changes").unwrap().to_owned();
- match FFLAGS.change_feed_live_queries.enabled() {
- true => {
- assert_eq!(
- changes,
- surrealdb::sql::value(
- "[
- {
- create: {
- id: user:amos,
- name: 'AMOS'
- }
- }
- ]"
- )
- .unwrap()
- );
- }
- false => {
- assert_eq!(
- changes,
- surrealdb::sql::value(
- "[
+ assert_eq!(
+ changes,
+ surrealdb::sql::value(
+ "[
  {
  update: {
  id: user:amos,
  name: 'AMOS'
  }
  }
  ]"
- )
- .unwrap()
- );
- }
- };
+ )
+ .unwrap()
+ );
+
  // UPDATE table
  let a = array.get(4).unwrap();
  let Value::Object(a) = a else {