chore: warehouse tests race detection (#3773)

* chore: warehouse tests race detection * temp * temp * temp * chore: fix defer * chore: removed unused lock * fix: unlocking of RWMutex * fix: unlocking of RWMutex * chore: fixing err data race * chore: removing parallel * chore: go-kit v0.15.6 * chore: removing uploadLock * chore: removing uploadSchema mutex * chore: removing TODO * chore: go-kit v0.15.7
rudderlabs · Aug 28, 2023 · ce2ed33 · ce2ed33
1 parent 890ccb6
commit ce2ed33
Show file tree

Hide file tree

Showing 11 changed files with 707 additions and 634 deletions.
diff --git a/Makefile b/Makefile
@@ -31,7 +31,7 @@ endif
 test-warehouse-integration:
 	$(eval TEST_PATTERN = 'TestIntegration')
 	$(eval TEST_CMD = SLOW=1 go test)
-	$(eval TEST_OPTIONS = -v -p 8 -timeout 30m -count 1 -run $(TEST_PATTERN) -coverprofile=profile.out -covermode=atomic -coverpkg=./...)
+	$(eval TEST_OPTIONS = -v -p 8 -timeout 30m -count 1 -race -run $(TEST_PATTERN) -coverprofile=profile.out -covermode=atomic -coverpkg=./...)
 	$(TEST_CMD) $(TEST_OPTIONS) $(package) && touch $(TESTFILE) || true
 
 test-warehouse: test-warehouse-integration test-teardown

diff --git a/warehouse/integrations/snowflake/snowflake_test.go b/warehouse/integrations/snowflake/snowflake_test.go
@@ -322,13 +322,16 @@ func TestIntegration(t *testing.T) {
 				cancel := bootstrap(t, tc.appendMode)
 				defer cancel()
 
+				cred := tc.cred
+				cred.Database = tc.database
+
 				urlConfig := sfdb.Config{
-					Account:   tc.cred.Account,
-					User:      tc.cred.User,
-					Role:      tc.cred.Role,
-					Password:  tc.cred.Password,
-					Database:  tc.database,
-					Warehouse: tc.cred.Warehouse,
+					Account:   cred.Account,
+					User:      cred.User,
+					Role:      cred.Role,
+					Password:  cred.Password,
+					Database:  cred.Database,
+					Warehouse: cred.Warehouse,
 				}
 
 				dsn, err := sfdb.DSN(&urlConfig)

diff --git a/warehouse/internal/model/schema.go b/warehouse/internal/model/schema.go
@@ -30,3 +30,14 @@ type WHSchema struct {
 	CreatedAt       time.Time
 	UpdatedAt       time.Time
 }
+
+func (s Schema) Clone() Schema {
+	cp := make(Schema, len(s))
+	for k, v := range s {
+		cp[k] = make(TableSchema, len(v))
+		for kk, vv := range v {
+			cp[k][kk] = vv
+		}
+	}
+	return cp
+}
diff --git a/warehouse/logfield/logfield.go b/warehouse/logfield/logfield.go
@@ -29,5 +29,6 @@ const (
 	QueryExecutionTime         = "queryExecutionTime"
 	StagingTableName           = "stagingTableName"
 	TotalRows                  = "totalRows"
+	WarehouseID                = "warehouseID"
 	SampleDuplicateMessages    = "sampleDuplicateMessages"
 )
diff --git a/warehouse/router_identities.go b/warehouse/router_identities.go
@@ -440,7 +440,7 @@ func (r *router) populateHistoricIdentities(ctx context.Context, warehouse model
 		}
 		defer whManager.Cleanup(ctx)
 
-		err = job.schemaHandle.fetchSchemaFromWarehouse(ctx, whManager)
+		_, _, err = job.schemaHandle.FetchSchemaFromWarehouse(ctx, whManager)
 		if err != nil {
 			r.logger.Errorf(`[WH]: Failed fetching schema from warehouse: %v`, err)
 			job.setUploadError(err, model.Aborted)