Add syntax for change feeds to store original value alongside change (#3567)

Co-authored-by: Mees Delzenne <DelSkayn@users.noreply.github.com>

core/src/cf/writer.rs

@@ -150,6 +150,7 @@ mod tests {
 			name: crate::sql::Ident(db.to_string()),
 			changefeed: Some(ChangeFeed {
 				expiry: Duration::from_secs(10),
+				store_original: false,
 			}),
 			..Default::default()
 		};
@@ -157,6 +158,7 @@ mod tests {
 			name: tb.into(),
 			changefeed: Some(ChangeFeed {
 				expiry: Duration::from_secs(10),
+				store_original: false,
 			}),
 			..Default::default()
 		};

core/src/sql/v1/arbitrary.rs

@@ -32,6 +32,7 @@ impl<'a> Arbitrary<'a> for ChangeFeed {
 	fn arbitrary(u: &mut Unstructured<'a>) -> Result<Self> {
 		Ok(Self {
 			expiry: time::Duration::new(u64::arbitrary(u)?, u32::arbitrary(u)?),
+			store_original: bool::arbitrary(u)?,
 		})
 	}
 }

core/src/sql/v1/change_feed_include.rs

@@ -0,0 +1,25 @@
+use revision::revisioned;
+use serde::{Deserialize, Serialize};
+use std::fmt;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Hash)]
+#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
+#[revisioned(revision = 1)]
+/// ChangeFeedInclude statements are an appendix
+pub enum ChangeFeedInclude {
+	Original,
+}
+
+impl Default for crate::sql::change_feed_include::ChangeFeedInclude {
+	fn default() -> Self {
+		Self::Original
+	}
+}
+
+impl fmt::Display for crate::sql::change_feed_include::ChangeFeedInclude {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		f.write_str(match self {
+			Self::Original => "Original",
+		})
+	}
+}

core/src/sql/v1/changefeed.rs

@@ -6,14 +6,19 @@ use std::str;
 use std::time;
 
 #[derive(Clone, Debug, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Hash)]
-#[revisioned(revision = 1)]
+#[revisioned(revision = 2)]
 pub struct ChangeFeed {
 	pub expiry: time::Duration,
+	#[revision(start = 2)]
+	pub store_original: bool,
 }
 
 impl Display for ChangeFeed {
 	fn fmt(&self, f: &mut Formatter) -> fmt::Result {
 		write!(f, "CHANGEFEED {}", Duration(self.expiry))?;
+		if self.store_original {
+			write!(f, " INCLUDE ORIGINAL")?;
+		};
 		Ok(())
 	}
 }
@@ -22,6 +27,7 @@ impl Default for ChangeFeed {
 	fn default() -> Self {
 		Self {
 			expiry: time::Duration::from_secs(0),
+			store_original: false,
 		}
 	}
 }

core/src/sql/v1/mod.rs

@@ -8,6 +8,7 @@ pub(crate) mod base;
 pub(crate) mod block;
 pub(crate) mod bytes;
 pub(crate) mod cast;
+pub(crate) mod change_feed_include;
 pub(crate) mod changefeed;
 pub(crate) mod cond;
 pub(crate) mod constant;

core/src/sql/v1/value/serde/ser/changefeed/mod.rs

@@ -38,6 +38,7 @@ impl ser::Serializer for Serializer {
 #[derive(Default)]
 pub struct SerializeChangeFeed {
 	expiry: Duration,
+	store_original: bool,
 }
 
 impl serde::ser::SerializeStruct for SerializeChangeFeed {
@@ -52,6 +53,9 @@ impl serde::ser::SerializeStruct for SerializeChangeFeed {
 			"expiry" => {
 				self.expiry = value.serialize(ser::duration::Serializer.wrap())?;
 			}
+			"store_original" => {
+				self.store_original = value.serialize(ser::primitive::bool::Serializer.wrap())?;
+			}
 			key => {
 				return Err(Error::custom(format!("unexpected field `ChangeFeed::{key}`")));
 			}
@@ -62,6 +66,7 @@ impl serde::ser::SerializeStruct for SerializeChangeFeed {
 	fn end(self) -> Result<Self::Ok, Error> {
 		Ok(ChangeFeed {
 			expiry: self.expiry,
+			store_original: self.store_original,
 		})
 	}
 }

core/src/sql/v2/arbitrary.rs

@@ -32,6 +32,7 @@ impl<'a> Arbitrary<'a> for ChangeFeed {
 	fn arbitrary(u: &mut Unstructured<'a>) -> Result<Self> {
 		Ok(Self {
 			expiry: time::Duration::new(u64::arbitrary(u)?, u32::arbitrary(u)?),
+			store_original: bool::arbitrary(u)?,
 		})
 	}
 }

core/src/sql/v2/change_feed_include.rs

@@ -0,0 +1,25 @@
+use revision::revisioned;
+use serde::{Deserialize, Serialize};
+use std::fmt;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Hash)]
+#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
+#[revisioned(revision = 1)]
+/// ChangeFeedInclude statements are an appendix
+pub enum ChangeFeedInclude {
+	Original,
+}
+
+impl Default for ChangeFeedInclude {
+	fn default() -> Self {
+		Self::Original
+	}
+}
+
+impl fmt::Display for ChangeFeedInclude {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		f.write_str(match self {
+			Self::Original => "Original",
+		})
+	}
+}

core/src/sql/v2/changefeed.rs

@@ -6,13 +6,18 @@ use std::str;
 use std::time;
 
 #[derive(Clone, Debug, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Hash)]
-#[revisioned(revision = 1)]
+#[revisioned(revision = 2)]
 pub struct ChangeFeed {
 	pub expiry: time::Duration,
+	#[revision(start = 2)]
+	pub store_original: bool,
 }
 impl Display for ChangeFeed {
 	fn fmt(&self, f: &mut Formatter) -> fmt::Result {
 		write!(f, "CHANGEFEED {}", Duration(self.expiry))?;
+		if self.store_original {
+			write!(f, " INCLUDE ORIGINAL")?;
+		};
 		Ok(())
 	}
 }
@@ -21,6 +26,7 @@ impl Default for ChangeFeed {
 	fn default() -> Self {
 		Self {
 			expiry: time::Duration::from_secs(0),
+			store_original: false,
 		}
 	}
 }

core/src/sql/v2/mod.rs

@@ -8,6 +8,7 @@ pub(crate) mod base;
 pub(crate) mod block;
 pub(crate) mod bytes;
 pub(crate) mod cast;
+pub(crate) mod change_feed_include;
 pub(crate) mod changefeed;
 pub(crate) mod cond;
 pub(crate) mod constant;

core/src/sql/v2/statements/define/table.rs

@@ -1,3 +1,9 @@
+use std::fmt::{self, Display, Write};
+
+use derive::Store;
+use revision::revisioned;
+use serde::{Deserialize, Serialize};
+
 use crate::ctx::Context;
 use crate::dbs::{Options, Transaction};
 use crate::doc::CursorDoc;
@@ -9,10 +15,6 @@ use crate::sql::{
 	statements::UpdateStatement,
 	Base, Ident, Permissions, Strand, Value, Values, View,
 };
-use derive::Store;
-use revision::revisioned;
-use serde::{Deserialize, Serialize};
-use std::fmt::{self, Display, Write};
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Store, Hash)]
 #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]

core/src/sql/v2/value/serde/ser/changefeed/mod.rs

@@ -38,6 +38,7 @@ impl ser::Serializer for Serializer {
 #[derive(Default)]
 pub struct SerializeChangeFeed {
 	expiry: Duration,
+	store_original: bool,
 }
 
 impl serde::ser::SerializeStruct for SerializeChangeFeed {
@@ -52,6 +53,9 @@ impl serde::ser::SerializeStruct for SerializeChangeFeed {
 			"expiry" => {
 				self.expiry = value.serialize(ser::duration::Serializer.wrap())?;
 			}
+			"store_original" => {
+				self.store_original = value.serialize(ser::primitive::bool::Serializer.wrap())?;
+			}
 			key => {
 				return Err(Error::custom(format!("unexpected field `ChangeFeed::{key}`")));
 			}
@@ -62,6 +66,7 @@ impl serde::ser::SerializeStruct for SerializeChangeFeed {
 	fn end(self) -> Result<Self::Ok, Error> {
 		Ok(ChangeFeed {
 			expiry: self.expiry,
+			store_original: self.store_original,
 		})
 	}
 }

core/src/syn/v1/part/mod.rs

@@ -67,10 +67,20 @@ pub fn changefeed(i: &str) -> IResult<&str, ChangeFeed> {
 	let (i, _) = tag_no_case("CHANGEFEED")(i)?;
 	let (i, _) = shouldbespace(i)?;
 	let (i, v) = cut(duration)(i)?;
+
+	let (i, store_original) = opt(|i| {
+		let (i, _) = shouldbespace(i)?;
+		let (i, _): (&str, &str) = tag_no_case("INCLUDE")(i)?;
+		let (i, _) = shouldbespace(i)?;
+		let (i, b): (&str, &str) = tag_no_case("ORIGINAL")(i)?;
+		Ok((i, b))
+	})(i)?;
+
 	Ok((
 		i,
 		ChangeFeed {
 			expiry: v.0,
+			store_original: store_original.is_some(),
 		},
 	))
 }
@@ -234,7 +244,6 @@ pub fn version(i: &str) -> IResult<&str, Version> {
 
 #[cfg(test)]
 mod tests {
-
 	use super::*;
 	use crate::sql::{Datetime, Idiom, Value};
 	use crate::syn::Parse;
@@ -256,7 +265,23 @@ mod tests {
 		assert_eq!(
 			out,
 			ChangeFeed {
-				expiry: time::Duration::from_secs(3600)
+				expiry: time::Duration::from_secs(3600),
+				store_original: false,
+			}
+		);
+	}
+
+	#[test]
+	fn changefeed_include_original() {
+		let sql = "CHANGEFEED 1h INCLUDE ORIGINAL";
+		let res = changefeed(sql);
+		let out = res.unwrap().1;
+		assert_eq!("CHANGEFEED 1h INCLUDE ORIGINAL", format!("{}", out));
+		assert_eq!(
+			out,
+			ChangeFeed {
+				expiry: time::Duration::from_secs(3600),
+				store_original: true,
 			}
 		);
 	}
@@ -343,7 +368,7 @@ mod tests {
 		let out = res.unwrap().1;
 		assert_eq!(
 			out,
-			Fetchs(vec![Fetch(Idiom::parse("field")), Fetch(Idiom::parse("other.field")),])
+			Fetchs(vec![Fetch(Idiom::parse("field")), Fetch(Idiom::parse("other.field"))])
 		);
 		assert_eq!("FETCH field, other.field", format!("{}", out));
 	}

core/src/syn/v2/lexer/keywords.rs

@@ -1,4 +1,5 @@
 use crate::{
+	sql::change_feed_include::ChangeFeedInclude,
 	sql::{language::Language, Algorithm},
 	syn::v2::token::{DistanceKind, Keyword, TokenKind},
 };
@@ -74,6 +75,7 @@ pub(crate) static KEYWORDS: phf::Map<UniCase<&'static str>, TokenKind> = phf_map
 	UniCase::ascii("GROUP") => TokenKind::Keyword(Keyword::Group),
 	UniCase::ascii("HIGHLIGHTS") => TokenKind::Keyword(Keyword::Highlights),
 	UniCase::ascii("IGNORE") => TokenKind::Keyword(Keyword::Ignore),
+	UniCase::ascii("INCLUDE") => TokenKind::Keyword(Keyword::Include),
 	UniCase::ascii("INDEX") => TokenKind::Keyword(Keyword::Index),
 	UniCase::ascii("INFO") => TokenKind::Keyword(Keyword::Info),
 	UniCase::ascii("INSERT") => TokenKind::Keyword(Keyword::Insert),
@@ -284,6 +286,9 @@ pub(crate) static KEYWORDS: phf::Map<UniCase<&'static str>, TokenKind> = phf_map
 	UniCase::ascii("MANHATTAN") => TokenKind::Distance(DistanceKind::Manhattan),
 	UniCase::ascii("HAMMING") => TokenKind::Distance(DistanceKind::Hamming),
 	UniCase::ascii("MINKOWSKI") => TokenKind::Distance(DistanceKind::Minkowski),
+
+	// Change Feed keywords
+	UniCase::ascii("ORIGINAL") => TokenKind::ChangeFeedInclude(ChangeFeedInclude::Original),
 };
 
 const fn jwks_token_kind() -> TokenKind {

core/src/syn/v2/parser/stmt/parts.rs

@@ -1,5 +1,6 @@
 //! Contains parsing code for smaller common parts of statements.
 
+use crate::sql::change_feed_include::ChangeFeedInclude;
 use crate::{
 	sql::{
 		changefeed::ChangeFeed, index::Distance, Base, Cond, Data, Duration, Fetch, Fetchs, Field,
@@ -331,8 +332,16 @@ impl Parser<'_> {
 	/// Expects the parser to have already eating the `CHANGEFEED` keyword
 	pub fn parse_changefeed(&mut self) -> ParseResult<ChangeFeed> {
 		let expiry = self.next_token_value::<Duration>()?.0;
+		let store_original = if self.eat(t!("INCLUDE")) {
+			expected!(self, TokenKind::ChangeFeedInclude(ChangeFeedInclude::Original));
+			true
+		} else {
+			false
+		};
+
 		Ok(ChangeFeed {
 			expiry,
+			store_original,
 		})
 	}
 

core/src/syn/v2/parser/test/stmt.rs

@@ -138,7 +138,9 @@ fn parse_define_namespace() {
 
 #[test]
 fn parse_define_database() {
-	let res = test_parse!(parse_stmt, "DEFINE DATABASE a COMMENT 'test' CHANGEFEED 10m").unwrap();
+	let res =
+		test_parse!(parse_stmt, "DEFINE DATABASE a COMMENT 'test' CHANGEFEED 10m INCLUDE ORIGINAL")
+			.unwrap();
 	assert_eq!(
 		res,
 		Statement::Define(DefineStatement::Database(DefineDatabaseStatement {
@@ -146,7 +148,8 @@ fn parse_define_database() {
 			name: Ident("a".to_string()),
 			comment: Some(Strand("test".to_string())),
 			changefeed: Some(ChangeFeed {
-				expiry: std::time::Duration::from_secs(60) * 10
+				expiry: std::time::Duration::from_secs(60) * 10,
+				store_original: true,
 			})
 		}))
 	);
@@ -277,7 +280,7 @@ fn parse_define_param() {
 #[test]
 fn parse_define_table() {
 	let res =
-		test_parse!(parse_stmt, r#"DEFINE TABLE name DROP SCHEMAFUL CHANGEFEED 1s PERMISSIONS FOR SELECT WHERE a = 1 AS SELECT foo FROM bar GROUP BY foo"#)
+		test_parse!(parse_stmt, r#"DEFINE TABLE name DROP SCHEMAFUL CHANGEFEED 1s INCLUDE ORIGINAL PERMISSIONS FOR SELECT WHERE a = 1 AS SELECT foo FROM bar GROUP BY foo"#)
 			.unwrap();
 
 	assert_eq!(
@@ -312,7 +315,8 @@ fn parse_define_table() {
 				delete: Permission::None,
 			},
 			changefeed: Some(ChangeFeed {
-				expiry: std::time::Duration::from_secs(1)
+				expiry: std::time::Duration::from_secs(1),
+				store_original: true,
 			}),
 			comment: None,
 		}))

core/src/syn/v2/parser/test/streaming.rs

@@ -163,6 +163,7 @@ fn statements() -> Vec<Statement> {
 			comment: Some(Strand("test".to_string())),
 			changefeed: Some(ChangeFeed {
 				expiry: std::time::Duration::from_secs(60) * 10,
+				store_original: false,
 			}),
 		})),
 		Statement::Define(DefineStatement::Database(DefineDatabaseStatement {
@@ -235,6 +236,7 @@ fn statements() -> Vec<Statement> {
 			},
 			changefeed: Some(ChangeFeed {
 				expiry: std::time::Duration::from_secs(1),
+				store_original: false,
 			}),
 			comment: None,
 		})),

core/src/syn/v2/token/keyword.rs

@@ -83,6 +83,7 @@ keyword! {
 	Group => "GROUP",
 	Highlights => "HIGHLIGHTS",
 	Ignore => "IGNORE",
+	Include => "INCLUDE",
 	Index => "INDEX",
 	Info => "INFO",
 	Insert => "INSERT",

core/src/syn/v2/token/mod.rs

@@ -8,6 +8,7 @@ pub use keyword::Keyword;
 mod mac;
 pub(crate) use mac::t;
 
+use crate::sql::change_feed_include::ChangeFeedInclude;
 use crate::sql::{language::Language, Algorithm};
 
 /// A location in the source passed to the lexer.
@@ -224,6 +225,7 @@ pub enum NumberKind {
 pub enum TokenKind {
 	Keyword(Keyword),
 	Algorithm(Algorithm),
+	ChangeFeedInclude(ChangeFeedInclude),
 	Language(Language),
 	Distance(DistanceKind),
 	Operator(Operator),
@@ -383,6 +385,7 @@ impl TokenKind {
 			TokenKind::At => "@",
 			TokenKind::Invalid => "Invalid",
 			TokenKind::Eof => "Eof",
+			TokenKind::ChangeFeedInclude(_) => "change feed include",
 		}
 	}
 }