set synced at for resync table

PeerDB-io · Nov 17, 2023 · a46292f · a46292f
1 parent 2b944a3
commit a46292f
Show file tree

Hide file tree

Showing 9 changed files with 1,899 additions and 549 deletions.
diff --git a/flow/connectors/snowflake/snowflake.go b/flow/connectors/snowflake/snowflake.go
@@ -1228,6 +1228,25 @@ func (c *SnowflakeConnector) RenameTables(req *protos.RenameTablesInput) (*proto
 		}
 	}()
 
+	if req.SyncedAtColName != nil {
+		for _, renameRequest := range req.RenameTableOptions {
+			resyncTblName := renameRequest.CurrentName
+
+			log.WithFields(log.Fields{
+				"flowName": req.FlowJobName,
+			}).Infof("setting synced at column for table '%s'...", resyncTblName)
+
+			activity.RecordHeartbeat(c.ctx, fmt.Sprintf("setting synced at column for table '%s'...",
+				resyncTblName))
+
+			_, err = renameTablesTx.ExecContext(c.ctx,
+				fmt.Sprintf("UPDATE %s SET %s = CURRENT_TIMESTAMP", resyncTblName, *req.SyncedAtColName))
+			if err != nil {
+				return nil, fmt.Errorf("unable to set synced at column for table %s: %w", resyncTblName, err)
+			}
+		}
+	}
+
 	if req.SoftDeleteColName != nil {
 		for _, renameRequest := range req.RenameTableOptions {
 			src := renameRequest.CurrentName

diff --git a/flow/generated/protos/flow.pb.go b/flow/generated/protos/flow.pb.go
diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go
@@ -235,6 +235,7 @@ func CDCFlowWorkflowWithConfig(
 			if cfg.SoftDelete {
 				renameOpts.SoftDeleteColName = &cfg.SoftDeleteColName
 			}
+			renameOpts.SyncedAtColName = &cfg.SyncedAtColName
 			correctedTableNameSchemaMapping := make(map[string]*protos.TableSchema)
 			for _, mapping := range cfg.TableMappings {
 				oldName := mapping.DestinationTableIdentifier

diff --git a/nexus/pt/src/peerdb_flow.rs b/nexus/pt/src/peerdb_flow.rs
@@ -122,6 +122,8 @@ pub struct RenameTablesInput {
     pub rename_table_options: ::prost::alloc::vec::Vec<RenameTableOption>,
     #[prost(string, optional, tag="4")]
     pub soft_delete_col_name: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(string, optional, tag="5")]
+    pub synced_at_col_name: ::core::option::Option<::prost::alloc::string::String>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

diff --git a/nexus/pt/src/peerdb_flow.serde.rs b/nexus/pt/src/peerdb_flow.serde.rs
@@ -4146,6 +4146,9 @@ impl serde::Serialize for RenameTablesInput {
         if self.soft_delete_col_name.is_some() {
             len += 1;
         }
+        if self.synced_at_col_name.is_some() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("peerdb_flow.RenameTablesInput", len)?;
         if !self.flow_job_name.is_empty() {
             struct_ser.serialize_field("flowJobName", &self.flow_job_name)?;
@@ -4159,6 +4162,9 @@ impl serde::Serialize for RenameTablesInput {
         if let Some(v) = self.soft_delete_col_name.as_ref() {
             struct_ser.serialize_field("softDeleteColName", v)?;
         }
+        if let Some(v) = self.synced_at_col_name.as_ref() {
+            struct_ser.serialize_field("syncedAtColName", v)?;
+        }
         struct_ser.end()
     }
 }
@@ -4176,6 +4182,8 @@ impl<'de> serde::Deserialize<'de> for RenameTablesInput {
             "renameTableOptions",
             "soft_delete_col_name",
             "softDeleteColName",
+            "synced_at_col_name",
+            "syncedAtColName",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -4184,6 +4192,7 @@ impl<'de> serde::Deserialize<'de> for RenameTablesInput {
             Peer,
             RenameTableOptions,
             SoftDeleteColName,
+            SyncedAtColName,
             __SkipField__,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
@@ -4210,6 +4219,7 @@ impl<'de> serde::Deserialize<'de> for RenameTablesInput {
                             "peer" => Ok(GeneratedField::Peer),
                             "renameTableOptions" | "rename_table_options" => Ok(GeneratedField::RenameTableOptions),
                             "softDeleteColName" | "soft_delete_col_name" => Ok(GeneratedField::SoftDeleteColName),
+                            "syncedAtColName" | "synced_at_col_name" => Ok(GeneratedField::SyncedAtColName),
                             _ => Ok(GeneratedField::__SkipField__),
                         }
                     }
@@ -4233,6 +4243,7 @@ impl<'de> serde::Deserialize<'de> for RenameTablesInput {
                 let mut peer__ = None;
                 let mut rename_table_options__ = None;
                 let mut soft_delete_col_name__ = None;
+                let mut synced_at_col_name__ = None;
                 while let Some(k) = map.next_key()? {
                     match k {
                         GeneratedField::FlowJobName => {
@@ -4259,6 +4270,12 @@ impl<'de> serde::Deserialize<'de> for RenameTablesInput {
                             }
                             soft_delete_col_name__ = map.next_value()?;
                         }
+                        GeneratedField::SyncedAtColName => {
+                            if synced_at_col_name__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("syncedAtColName"));
+                            }
+                            synced_at_col_name__ = map.next_value()?;
+                        }
                         GeneratedField::__SkipField__ => {
                             let _ = map.next_value::<serde::de::IgnoredAny>()?;
                         }
@@ -4269,6 +4286,7 @@ impl<'de> serde::Deserialize<'de> for RenameTablesInput {
                     peer: peer__,
                     rename_table_options: rename_table_options__.unwrap_or_default(),
                     soft_delete_col_name: soft_delete_col_name__,
+                    synced_at_col_name: synced_at_col_name__,
                 })
             }
         }

diff --git a/protos/flow.proto b/protos/flow.proto
@@ -84,6 +84,7 @@ message RenameTablesInput {
   peerdb_peers.Peer peer = 2;
   repeated RenameTableOption rename_table_options = 3;
   optional string soft_delete_col_name = 4;
+  optional string synced_at_col_name = 5;
 }
 
 message RenameTablesOutput {

diff --git a/ui/grpc_generated/flow.ts b/ui/grpc_generated/flow.ts
@@ -165,6 +165,7 @@ export interface RenameTablesInput {
   peer: Peer | undefined;
   renameTableOptions: RenameTableOption[];
   softDeleteColName?: string | undefined;
+  syncedAtColName?: string | undefined;
 }
 
 export interface RenameTablesOutput {
@@ -1592,7 +1593,13 @@ export const RenameTableOption = {
 };
 
 function createBaseRenameTablesInput(): RenameTablesInput {
-  return { flowJobName: "", peer: undefined, renameTableOptions: [], softDeleteColName: undefined };
+  return {
+    flowJobName: "",
+    peer: undefined,
+    renameTableOptions: [],
+    softDeleteColName: undefined,
+    syncedAtColName: undefined,
+  };
 }
 
 export const RenameTablesInput = {
@@ -1609,6 +1616,9 @@ export const RenameTablesInput = {
     if (message.softDeleteColName !== undefined) {
       writer.uint32(34).string(message.softDeleteColName);
     }
+    if (message.syncedAtColName !== undefined) {
+      writer.uint32(42).string(message.syncedAtColName);
+    }
     return writer;
   },
 
@@ -1647,6 +1657,13 @@ export const RenameTablesInput = {
 
           message.softDeleteColName = reader.string();
           continue;
+        case 5:
+          if (tag !== 42) {
+            break;
+          }
+
+          message.syncedAtColName = reader.string();
+          continue;
       }
       if ((tag & 7) === 4 || tag === 0) {
         break;
@@ -1664,6 +1681,7 @@ export const RenameTablesInput = {
         ? object.renameTableOptions.map((e: any) => RenameTableOption.fromJSON(e))
         : [],
       softDeleteColName: isSet(object.softDeleteColName) ? String(object.softDeleteColName) : undefined,
+      syncedAtColName: isSet(object.syncedAtColName) ? String(object.syncedAtColName) : undefined,
     };
   },
 
@@ -1681,6 +1699,9 @@ export const RenameTablesInput = {
     if (message.softDeleteColName !== undefined) {
       obj.softDeleteColName = message.softDeleteColName;
     }
+    if (message.syncedAtColName !== undefined) {
+      obj.syncedAtColName = message.syncedAtColName;
+    }
     return obj;
   },
 
@@ -1693,6 +1714,7 @@ export const RenameTablesInput = {
     message.peer = (object.peer !== undefined && object.peer !== null) ? Peer.fromPartial(object.peer) : undefined;
     message.renameTableOptions = object.renameTableOptions?.map((e) => RenameTableOption.fromPartial(e)) || [];
     message.softDeleteColName = object.softDeleteColName ?? undefined;
+    message.syncedAtColName = object.syncedAtColName ?? undefined;
     return message;
   },
 };