Compare Convex schemas to a Fivetran destination (#26916)

GitOrigin-RevId: 939df3e2070e85b25de01b59aa1922f8c20af7a5

Author: Nicolapps

Cargo.lock

@@ -39,6 +39,7 @@ tonic = { workspace = true, features = ["gzip"] }
 [dev-dependencies]
 common = { path = "../common", features = ["testing"] }
 convex_fivetran_common = { path = "../fivetran_common", features = ["testing"] }
+must-let = { workspace = true }
 proptest = { workspace = true }
 
 [package.metadata.cargo-machete]

crates/fivetran_destination/Cargo.toml

@@ -7,12 +7,47 @@ use chrono::{
     DateTime,
     Utc,
 };
-use common::value::ConvexObject;
+use common::value::{
+    ConvexObject,
+    FieldPath,
+    IdentifierFieldName,
+};
 use serde::{
     Deserialize,
     Serialize,
 };
 
+use crate::constants::{
+    ID_FIELD_PATH,
+    ID_FIVETRAN_FIELD_NAME,
+    SOFT_DELETE_FIELD_PATH,
+    SOFT_DELETE_FIVETRAN_FIELD_NAME,
+    SYNCED_FIELD_PATH,
+    SYNCED_FIVETRAN_FIELD_NAME,
+};
+
+#[derive(
+    Hash, Eq, PartialEq, derive_more::Display, Debug, serde::Deserialize, Clone, PartialOrd, Ord,
+)]
+#[serde(transparent)]
+pub struct FivetranTableName(String);
+
+impl FromStr for FivetranTableName {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(Self(s.to_owned()))
+    }
+}
+
+impl Deref for FivetranTableName {
+    type Target = str;
+
+    fn deref(&self) -> &str {
+        &self.0[..]
+    }
+}
+
 #[derive(
     Hash, Eq, PartialEq, derive_more::Display, Debug, serde::Deserialize, Clone, PartialOrd, Ord,
 )]
@@ -35,6 +70,23 @@ impl Deref for FivetranFieldName {
     }
 }
 
+impl TryInto<FieldPath> for FivetranFieldName {
+    type Error = anyhow::Error;
+
+    fn try_into(self) -> Result<FieldPath, Self::Error> {
+        Ok(if &self == SYNCED_FIVETRAN_FIELD_NAME.deref() {
+            SYNCED_FIELD_PATH.clone()
+        } else if &self == SOFT_DELETE_FIVETRAN_FIELD_NAME.deref() {
+            SOFT_DELETE_FIELD_PATH.clone()
+        } else if &self == ID_FIVETRAN_FIELD_NAME.deref() {
+            ID_FIELD_PATH.clone()
+        } else {
+            let field = IdentifierFieldName::from_str(&self)?;
+            FieldPath::for_root_field(field)
+        })
+    }
+}
+
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 pub enum BatchWriteOperation {
     Upsert,
@@ -62,3 +114,42 @@ pub struct TruncateTableArgs {
     pub delete_before: Option<DateTime<Utc>>,
     pub delete_type: DeleteType,
 }
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use common::value::FieldPath;
+
+    use crate::api_types::FivetranFieldName;
+
+    #[test]
+    fn convert_fivetran_user_fields_to_field_path() {
+        let expected: FieldPath = FivetranFieldName::from_str("name")
+            .unwrap()
+            .try_into()
+            .unwrap();
+        assert_eq!(expected, FieldPath::from_str("name").unwrap());
+    }
+
+    #[test]
+    fn convert_fivetran_metadata_fields_to_field_path() {
+        let expected: FieldPath = FivetranFieldName::from_str("_fivetran_synced")
+            .unwrap()
+            .try_into()
+            .unwrap();
+        assert_eq!(expected, FieldPath::from_str("fivetran.synced").unwrap());
+
+        let expected: FieldPath = FivetranFieldName::from_str("_fivetran_id")
+            .unwrap()
+            .try_into()
+            .unwrap();
+        assert_eq!(expected, FieldPath::from_str("fivetran.id").unwrap());
+
+        let expected: FieldPath = FivetranFieldName::from_str("_fivetran_deleted")
+            .unwrap()
+            .try_into()
+            .unwrap();
+        assert_eq!(expected, FieldPath::from_str("fivetran.deleted").unwrap());
+    }
+}

crates/fivetran_destination/src/api_types.rs

@@ -1,6 +1,8 @@
 use std::sync::LazyLock;
 
 use common::{
+    bootstrap_model::index::database_index::IndexedFields,
+    document::CREATION_TIME_FIELD_PATH,
     types::IndexDescriptor,
     value::{
         FieldPath,
@@ -46,3 +48,30 @@ pub static SYNCED_FIELD_PATH: LazyLock<FieldPath> = LazyLock::new(|| {
     ])
     .expect("Invalid field path")
 });
+
+pub static ID_FIELD_PATH: LazyLock<FieldPath> = LazyLock::new(|| {
+    FieldPath::new(vec![
+        METADATA_CONVEX_FIELD_NAME.clone(),
+        ID_CONVEX_FIELD_NAME.clone(),
+    ])
+    .expect("Invalid field path")
+});
+
+pub static FIVETRAN_SYNC_INDEX_WITHOUT_SOFT_DELETE_FIELDS: LazyLock<IndexedFields> =
+    LazyLock::new(|| {
+        IndexedFields::try_from(vec![
+            SYNCED_FIELD_PATH.clone(),
+            CREATION_TIME_FIELD_PATH.clone(),
+        ])
+        .expect("Invalid IndexedFields")
+    });
+
+pub static FIVETRAN_SYNC_INDEX_WITH_SOFT_DELETE_FIELDS: LazyLock<IndexedFields> =
+    LazyLock::new(|| {
+        IndexedFields::try_from(vec![
+            SOFT_DELETE_FIELD_PATH.clone(),
+            SYNCED_FIELD_PATH.clone(),
+            CREATION_TIME_FIELD_PATH.clone(),
+        ])
+        .expect("Invalid IndexedFields")
+    });