fix: warehouse sync job queries for snowflake and bigquery

warehouse/bigquery/bigquery.go

@@ -239,14 +239,19 @@ func (bq *HandleT) dropStagingTable(stagingTableName string) {
 
 func (bq *HandleT) DeleteBy(tableNames []string, params warehouseutils.DeleteByParams) error {
  pkgLogger.Infof("BQ: Cleaning up the following tables in bigquery for BQ:%s : %v", tableNames)
+
  for _, tb := range tableNames {
- sqlStatement := fmt.Sprintf(`DELETE FROM "%[1]s"."%[2]s" WHERE
- context_sources_job_run_id <> @jobrunid AND
- context_sources_task_run_id <> @taskrunid AND
- context_source_id = @sourceid AND
- received_at < @starttime`,
- bq.namespace,
- tb,
+ tableName := fmt.Sprintf("`%s`.`%s`", bq.namespace, tb)
+ sqlStatement := fmt.Sprintf(`
+ DELETE FROM
+ %[1]s
+ WHERE
+ context_sources_job_run_id <> @jobrunid AND
+ context_sources_task_run_id <> @taskrunid AND
+ context_source_id = @sourceid AND
+ received_at < @starttime;
+ `,
+ tableName,
  )
 
  pkgLogger.Infof("PG: Deleting rows in table in bigquery for BQ:%s", bq.warehouse.Destination.ID)
@@ -273,7 +278,6 @@ func (bq *HandleT) DeleteBy(tableNames []string, params warehouseutils.DeleteByP
  return status.Err()
  }
  }
-
  }
  return nil
 }

warehouse/jobs/handlers.go

@@ -15,6 +15,7 @@ import (
  "encoding/json"
  "io"
  "net/http"
+ "strings"
 )
 
 // AddWarehouseJobHandler The following handler gets called for adding async
@@ -51,35 +52,40 @@ func (a *AsyncJobWhT) AddWarehouseJobHandler(w http.ResponseWriter, r *http.Requ
  http.Error(w, "Error extracting tableNames", http.StatusBadRequest)
  return
  }
+
  var jobIds []int64
  // Add to wh_async_job queue each of the tables
  for _, th := range tableNames {
- if !skipTable(th) {
- jobsMetaData := WhJobsMetaData{
- JobRunID: startJobPayload.JobRunID,
- TaskRunID: startJobPayload.TaskRunID,
- StartTime: startJobPayload.StartTime,
- JobType: AsyncJobType,
- }
- metadataJson, err := json.Marshal(jobsMetaData)
- if err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
- return
- }
- payload := AsyncJobPayloadT{
- SourceID: startJobPayload.SourceID,
- DestinationID: startJobPayload.DestinationID,
- TableName: th,
- AsyncJobType: startJobPayload.AsyncJobType,
- MetaData: metadataJson,
- }
- id, err := a.addJobsToDB(a.context, &payload)
- if err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
- return
- }
- jobIds = append(jobIds, id)
+
+ switch strings.ToLower(th) {
+ case "rudder_discards", "rudder_identity_mappings", "rudder_identity_merge_rules":
+ continue
+ }
+
+ jobsMetaData := WhJobsMetaData{
+ JobRunID: startJobPayload.JobRunID,
+ TaskRunID: startJobPayload.TaskRunID,
+ StartTime: startJobPayload.StartTime,
+ JobType: AsyncJobType,
+ }
+ metadataJson, err := json.Marshal(jobsMetaData)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ payload := AsyncJobPayloadT{
+ SourceID: startJobPayload.SourceID,
+ DestinationID: startJobPayload.DestinationID,
+ TableName: th,
+ AsyncJobType: startJobPayload.AsyncJobType,
+ MetaData: metadataJson,
+ }
+ id, err := a.addJobsToDB(a.context, &payload)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
  }
+ jobIds = append(jobIds, id)
  }
  whAddJobResponse := WhAddJobResponse{
  JobIds: jobIds,

warehouse/jobs/utils.go

@@ -38,13 +38,6 @@ func validatePayload(payload StartJobReqPayload) bool {
  return true
 }
 
-func skipTable(th string) bool {
- if th == "RUDDER_DISCARDS" || th == "rudder_discards" {
- return true
- }
- return false
-}
-
 func contains(sArray []string, s string) bool {
  for _, s1 := range sArray {
  if s1 == s {

warehouse/snowflake/snowflake.go

@@ -214,25 +214,28 @@ func (sf *HandleT) authString() string {
 func (sf *HandleT) DeleteBy(tableNames []string, params warehouseutils.DeleteByParams) (err error) {
  pkgLogger.Infof("SF: Cleaning up the following tables in snowflake for SF:%s : %v", tableNames)
  for _, tb := range tableNames {
- sqlStatement := fmt.Sprintf(`DELETE FROM "%[1]s"."%[2]s" WHERE
- context_sources_job_run_id <> :jobrunid AND
- context_sources_task_run_id <> :taskrunid AND
- context_source_id = :sourceid AND
- received_at < :starttime`,
+ sqlStatement := fmt.Sprintf(`
+ DELETE FROM
+ %[1]q.%[2]q
+ WHERE
+ context_sources_job_run_id <> '%[3]s' AND
+ context_sources_task_run_id <> '%[4]s' AND
+ context_source_id = '%[5]s' AND
+ received_at < '%[6]s';
+ `,
  sf.Namespace,
  tb,
+ params.JobRunId,
+ params.TaskRunId,
+ params.SourceId,
+ params.StartTime,
  )
 
  pkgLogger.Infof("SF: Deleting rows in table in snowflake for SF:%s", sf.Warehouse.Destination.ID)
  pkgLogger.Debugf("SF: Executing the sql statement %v", sqlStatement)
 
  if enableDeleteByJobs {
- _, err = sf.Db.Exec(sqlStatement,
- sql.Named("jobrunid", params.JobRunId),
- sql.Named("taskrunid", params.TaskRunId),
- sql.Named("sourceid", params.SourceId),
- sql.Named("starttime", params.StartTime),
- )
+ _, err = sf.Db.Exec(sqlStatement)
  if err != nil {
  pkgLogger.Errorf("Error %s", err)
  return err