adds fix for pg, adds comments

PeerDB-io · Dec 21, 2023 · eac5b85 · eac5b85
1 parent f78bec3
commit eac5b85
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 1 deletion.
diff --git a/flow/connectors/bigquery/merge_statement_generator.go b/flow/connectors/bigquery/merge_statement_generator.go
@@ -229,6 +229,10 @@ func (m *mergeStmtGenerator) generateUpdateStatements(
 		(_peerdb_deduped._peerdb_record_type != 2) AND _peerdb_unchanged_toast_columns='%s'
 		THEN UPDATE SET %s `, cols, ssep)
 		updateStmts = append(updateStmts, updateStmt)
+
+		// generates update statements for the case where updates and deletes happen in the same branch
+		// the backfill has happened from the pull side already, so treat the DeleteRecord as an update
+		// and then set soft-delete to true.
 		if peerdbCols.SoftDelete && (peerdbCols.SoftDeleteColName != "") {
 			tmpArray = append(tmpArray[:len(tmpArray)-1],
 				fmt.Sprintf("`%s` = TRUE", peerdbCols.SoftDeleteColName))

diff --git a/flow/connectors/postgres/client.go b/flow/connectors/postgres/client.go
@@ -761,7 +761,7 @@ func (c *PostgresConnector) generateUpdateStatement(allCols []string,
 				peerdbCols.SyncedAtColName))
 		}
 		// set soft-deleted to false, tackles insert after soft-delete
-		if peerdbCols.SoftDeleteColName != "" {
+		if peerdbCols.SoftDelete && (peerdbCols.SoftDeleteColName != "") {
 			tmpArray = append(tmpArray, fmt.Sprintf(`"%s" = FALSE`,
 				peerdbCols.SoftDeleteColName))
 		}
@@ -771,6 +771,19 @@ func (c *PostgresConnector) generateUpdateStatement(allCols []string,
 			src._peerdb_record_type=1 AND _peerdb_unchanged_toast_columns='%s'
 			THEN UPDATE SET %s `, cols, ssep)
 		updateStmts = append(updateStmts, updateStmt)
+
+		// generates update statements for the case where updates and deletes happen in the same branch
+		// the backfill has happened from the pull side already, so treat the DeleteRecord as an update
+		// and then set soft-delete to true.
+		if peerdbCols.SoftDelete && (peerdbCols.SoftDeleteColName != "") {
+			tmpArray = append(tmpArray[:len(tmpArray)-1],
+				fmt.Sprintf(`"%s" = TRUE`, peerdbCols.SoftDeleteColName))
+			ssep := strings.Join(tmpArray, ", ")
+			updateStmt := fmt.Sprintf(`WHEN MATCHED AND
+			src._peerdb_record_type = 2 AND _peerdb_unchanged_toast_columns='%s'
+			THEN UPDATE SET %s `, cols, ssep)
+			updateStmts = append(updateStmts, updateStmt)
+		}
 	}
 	return updateStmts
 }

diff --git a/flow/connectors/snowflake/snowflake.go b/flow/connectors/snowflake/snowflake.go
@@ -1064,6 +1064,10 @@ func (c *SnowflakeConnector) generateUpdateStatements(
 		(SOURCE._PEERDB_RECORD_TYPE != 2) AND _PEERDB_UNCHANGED_TOAST_COLUMNS='%s'
 		THEN UPDATE SET %s `, cols, ssep)
 		updateStmts = append(updateStmts, updateStmt)
+
+		// generates update statements for the case where updates and deletes happen in the same branch
+		// the backfill has happened from the pull side already, so treat the DeleteRecord as an update
+		// and then set soft-delete to true.
 		if softDelete && (softDeleteCol != "") {
 			tmpArray = append(tmpArray[:len(tmpArray)-1], fmt.Sprintf(`"%s" = TRUE`, softDeleteCol))
 			ssep := strings.Join(tmpArray, ", ")