fix rebase

Author: Amogh-Bharadwaj

flow/connectors/postgres/client.go

@@ -55,12 +55,14 @@ const (
 	ARRAY_AGG(DISTINCT _peerdb_unchanged_toast_columns) FROM %s.%s WHERE
 	_peerdb_batch_id=$1 AND _peerdb_record_type!=2 GROUP BY _peerdb_destination_table_name`
 	mergeStatementSQL = `WITH src_rank AS (
-		SELECT _peerdb_data,_peerdb_record_type,_peerdb_unchanged_toast_columns,_peerdb_timestamp,
+		SELECT r.*,_peerdb_record_type,_peerdb_unchanged_toast_columns,_peerdb_timestamp,
 		RANK() OVER (PARTITION BY %s ORDER BY _peerdb_timestamp DESC) AS _peerdb_rank
-		FROM %s.%s WHERE _peerdb_batch_id = $1 AND _peerdb_destination_table_name=$2
+		FROM %s.%s, jsonb_to_record(_peerdb_data) AS r(%s)
+		WHERE _peerdb_batch_id = $1 AND _peerdb_destination_table_name = $2
 	)
 	MERGE INTO %s dst
-	USING (SELECT %s,_peerdb_record_type,_peerdb_unchanged_toast_columns FROM src_rank WHERE _peerdb_rank=1 ORDER BY _peerdb_timestamp) src
+	USING (SELECT %s,_peerdb_record_type,_peerdb_unchanged_toast_columns
+		FROM src_rank WHERE _peerdb_rank=1 ORDER BY _peerdb_timestamp) src
 	ON %s
 	WHEN NOT MATCHED AND src._peerdb_record_type!=2 THEN
 	INSERT (%s) VALUES (%s) %s