fix: remove source labels in repo query stats (#6483)

🔒 Scanned for secrets using gitleaks 8.28.0

# Description

- Remove sourceId labels from warehouse repo query stats.

## Linear Ticket

- Resolves INT-4323

## Security

- [x] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Removes sourceId labels from warehouse repo query stats and downgrades
a snowpipe streaming polling log line to debug.
> 
> - **Warehouse Repo (stats tags)**:
> - Drop `sourceId` from `TimerStat` tags across repos: `load.go`
(`Insert`, `DistinctTableName`), `schema.go` (`Insert`, `GetNamespace`),
`source.go` (`Insert`), `staging.go` (`Insert`, `Pending`,
`CountPendingForSource`), `staging_snapshot.go` (`Insert`, `GetLatest`),
`table_upload.go` (`GetByJobRunTaskRun`), `upload.go` (`LastCreatedAt`,
`GetLatestUploadInfo`, `RetryFailedBatches`, `GetSyncLatencies`).
> - Update tests in `warehouse/internal/repo/repo_test.go` to align with
removed `sourceId` tags.
> - **Snowpipe Streaming**:
> - Change polling log for in-progress flushing from `Infon` to `Debugn`
in `snowpipestreaming.go`.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
4bdd695. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: achettyiitr

router/batchrouter/asyncdestinationmanager/snowpipestreaming/snowpipestreaming.go

@@ -676,7 +676,7 @@ func isInProgress(statusRes *model.StatusResponse, info *importInfo, log logger.
 
 	// Case 3: Flushing in progress - continue polling
 	if latestInsertedOffset > latestCommittedOffset {
-		log.Infon("Flushing in progress, continuing to poll")
+		log.Debugn("Flushing in progress, continuing to poll")
 		return true, nil
 	}
 

warehouse/internal/repo/load.go

@@ -76,7 +76,6 @@ func (lf *LoadFiles) Delete(ctx context.Context, uploadID int64) error {
 // Insert loadFiles into the database.
 func (lf *LoadFiles) Insert(ctx context.Context, loadFiles []model.LoadFile) error {
 	defer lf.TimerStat("insert", stats.Tags{
-		"sourceId": loadFiles[0].SourceID,
 		"destId":   loadFiles[0].DestinationID,
 		"destType": loadFiles[0].DestinationType,
 	})()
@@ -265,8 +264,7 @@ func (lf *LoadFiles) DistinctTableName(
 	endID int64,
 ) ([]string, error) {
 	defer lf.TimerStat("distinct_table_name", stats.Tags{
-		"sourceId": sourceID,
-		"destId":   destinationID,
+		"destId": destinationID,
 	})()
 
 	rows, err := lf.db.QueryContext(ctx, `

warehouse/internal/repo/repo_test.go

@@ -61,7 +61,6 @@ func TestStatsEmission(t *testing.T) {
 			},
 		}))
 		require.Greater(t, statsStore.Get("warehouse_repo_query_wh_load_files_insert_duration_seconds", stats.Tags{
-			"sourceId": "source_id",
 			"destId":   "destination_id",
 			"destType": "destination_type",
 		}).LastDuration(), time.Duration(0))
@@ -74,15 +73,13 @@ func TestStatsEmission(t *testing.T) {
 		})
 		require.NoError(t, err)
 		require.Greater(t, statsStore.Get("warehouse_repo_query_wh_schemas_insert_duration_seconds", stats.Tags{
-			"sourceId": "source_id",
 			"destId":   "destination_id",
 			"destType": "destination_type",
 		}).LastDuration(), time.Duration(0))
 
 		_, err = repoStagingFileSchemaSnapshots.Insert(ctx, "source_id", "destination_id", "workspace_id", json.RawMessage(`{}`))
 		require.NoError(t, err)
 		require.Greater(t, statsStore.Get("warehouse_repo_query_wh_staging_file_schema_snapshots_insert_duration_seconds", stats.Tags{
-			"sourceId":    "source_id",
 			"destId":      "destination_id",
 			"workspaceId": "workspace_id",
 		}).LastDuration(), time.Duration(0))
@@ -94,7 +91,6 @@ func TestStatsEmission(t *testing.T) {
 		}).WithSchema(json.RawMessage(`{}`))))
 		require.NoError(t, err)
 		require.Greater(t, statsStore.Get("warehouse_repo_query_wh_staging_files_insert_duration_seconds", stats.Tags{
-			"sourceId":    "source_id",
 			"destId":      "destination_id",
 			"workspaceId": "workspace_id",
 		}).LastDuration(), time.Duration(0))
@@ -115,13 +111,11 @@ func TestStatsEmission(t *testing.T) {
 		})
 		require.NoError(t, err)
 		require.Greater(t, statsStore.Get("warehouse_repo_query_wh_async_jobs_insert_duration_seconds", stats.Tags{
-			"sourceId":    "source_id",
 			"destId":      "destination_id",
 			"workspaceId": "workspace_id",
 		}).LastDuration(), time.Duration(0))
 
 		require.Greater(t, statsStore.Get("warehouse_repo_query_wh_uploads_create_with_staging_files_duration_seconds", stats.Tags{
-			"sourceId":    "source_id",
 			"destId":      "destination_id",
 			"destType":    "destination_type",
 			"workspaceId": "workspace_id",

warehouse/internal/repo/schema.go

@@ -72,7 +72,6 @@ func NewWHSchemas(db *sqlmiddleware.DB, conf *config.Config, log logger.Logger,
 // If Warehouse.enableTableLevelSchema is true in config, it also inserts/updates table-level schemas for each table in the schema.
 func (sh *WHSchema) Insert(ctx context.Context, whSchema *model.WHSchema) error {
 	defer sh.TimerStat("insert", stats.Tags{
-		"sourceId": whSchema.SourceID,
 		"destId":   whSchema.DestinationID,
 		"destType": whSchema.DestinationType,
 	})()
@@ -478,8 +477,7 @@ func parseWHSchemas(rows *sqlmiddleware.Rows) ([]*model.WHSchema, error) {
 
 func (sh *WHSchema) GetNamespace(ctx context.Context, sourceID, destID string) (string, error) {
 	defer sh.TimerStat("get_namespace", stats.Tags{
-		"sourceId": sourceID,
-		"destId":   destID,
+		"destId": destID,
 	})()
 
 	query := `SELECT namespace FROM ` + whSchemaTableName + `

warehouse/internal/repo/source.go

@@ -54,7 +54,6 @@ func (s *Source) Insert(ctx context.Context, sourceJobs []model.SourceJob) ([]in
 		return nil, errors.New("empty sourceJobs")
 	}
 	defer (*repo)(s).TimerStat("insert", stats.Tags{
-		"sourceId":    sourceJobs[0].SourceID,
 		"destId":      sourceJobs[0].DestinationID,
 		"workspaceId": sourceJobs[0].WorkspaceID,
 	})()

warehouse/internal/repo/staging.go

@@ -157,7 +157,6 @@ func (sf *StagingFiles) Insert(ctx context.Context, stagingFile *model.StagingFi
 	}
 
 	defer sf.TimerStat("insert", stats.Tags{
-		"sourceId":    stagingFile.SourceID,
 		"destId":      stagingFile.DestinationID,
 		"workspaceId": stagingFile.WorkspaceID,
 	})()
@@ -391,8 +390,7 @@ func (sf *StagingFiles) GetForUploadID(ctx context.Context, uploadID int64) ([]*
 
 func (sf *StagingFiles) Pending(ctx context.Context, sourceID, destinationID string) ([]*model.StagingFile, error) {
 	defer sf.TimerStat("pending", stats.Tags{
-		"sourceId": sourceID,
-		"destId":   destinationID,
+		"destId": destinationID,
 	})()
 
 	var (
@@ -437,9 +435,7 @@ func (sf *StagingFiles) Pending(ctx context.Context, sourceID, destinationID str
 }
 
 func (sf *StagingFiles) CountPendingForSource(ctx context.Context, sourceID string) (int64, error) {
-	defer sf.TimerStat("count_pending_for_source", stats.Tags{
-		"sourceId": sourceID,
-	})()
+	defer sf.TimerStat("count_pending_for_source", nil)()
 
 	return sf.countPending(ctx, `source_id = $1`, sourceID)
 }

warehouse/internal/repo/staging_snapshot.go

@@ -44,7 +44,6 @@ func NewStagingFileSchemaSnapshots(db *sqlmiddleware.DB, opts ...Opt) *StagingFi
 // Insert inserts a new schema snapshot into the database and returns its auto-generated ID.
 func (r *StagingFileSchemaSnapshots) Insert(ctx context.Context, sourceID, destinationID, workspaceID string, schemaBytes json.RawMessage) (uuid.UUID, error) {
 	defer (*repo)(r).TimerStat("insert", stats.Tags{
-		"sourceId":    sourceID,
 		"destId":      destinationID,
 		"workspaceId": workspaceID,
 	})()
@@ -75,8 +74,7 @@ func (r *StagingFileSchemaSnapshots) Insert(ctx context.Context, sourceID, desti
 // Returns ErrNoSchemaSnapshot if not found.
 func (r *StagingFileSchemaSnapshots) GetLatest(ctx context.Context, sourceID, destinationID string) (*model.StagingFileSchemaSnapshot, error) {
 	defer (*repo)(r).TimerStat("get_latest", stats.Tags{
-		"sourceId": sourceID,
-		"destId":   destinationID,
+		"destId": destinationID,
 	})()
 
 	query := `

warehouse/internal/repo/table_upload.go

@@ -413,8 +413,7 @@ func (tu *TableUploads) GetByJobRunTaskRun(
 	taskRunID string,
 ) ([]model.TableUpload, error) {
 	defer tu.TimerStat("get_by_job_run_task_run", stats.Tags{
-		"sourceId": sourceID,
-		"destId":   destinationID,
+		"destId": destinationID,
 	})()
 
 	rows, err := tu.db.QueryContext(ctx, `

warehouse/internal/repo/upload.go

@@ -156,7 +156,6 @@ func (u *Uploads) CreateWithStagingFiles(ctx context.Context, upload model.Uploa
 	}
 
 	defer (*repo)(u).TimerStat("create_with_staging_files", stats.Tags{
-		"sourceId":    upload.SourceID,
 		"destId":      upload.DestinationID,
 		"destType":    upload.DestinationType,
 		"workspaceId": upload.WorkspaceID,
@@ -659,8 +658,7 @@ func (u *Uploads) ResetInProgress(ctx context.Context, destType string) error {
 
 func (u *Uploads) LastCreatedAt(ctx context.Context, sourceID, destinationID string) (time.Time, error) {
 	defer (*repo)(u).TimerStat("last_created_at", stats.Tags{
-		"sourceId": sourceID,
-		"destId":   destinationID,
+		"destId": destinationID,
 	})()
 
 	row := u.db.QueryRowContext(ctx, `
@@ -1068,8 +1066,7 @@ func (u *Uploads) RetryCount(ctx context.Context, opts model.RetryOptions) (int6
 
 func (u *Uploads) GetLatestUploadInfo(ctx context.Context, sourceID, destinationID string) (*model.LatestUploadInfo, error) {
 	defer (*repo)(u).TimerStat("get_latest_upload_info", stats.Tags{
-		"sourceId": sourceID,
-		"destId":   destinationID,
+		"destId": destinationID,
 	})()
 
 	var latestUploadInfo model.LatestUploadInfo
@@ -1309,7 +1306,6 @@ func (u *Uploads) RetryFailedBatches(
 	req model.RetryFailedBatchesRequest,
 ) (int64, error) {
 	defer (*repo)(u).TimerStat("retry_failed_batches", stats.Tags{
-		"sourceId":    req.SourceID,
 		"destId":      req.DestinationID,
 		"workspaceId": req.WorkspaceID,
 		"status":      req.Status,
@@ -1479,7 +1475,6 @@ func (u *Uploads) GetSyncLatencies(ctx context.Context, request model.SyncLatenc
 	defer (*repo)(u).TimerStat("get_sync_latencies", stats.Tags{
 		"destId":      request.DestinationID,
 		"workspaceId": request.WorkspaceID,
-		"sourceId":    request.SourceID,
 	})()
 
 	aggregationSQL := getLatencyAggregationSQL(request.AggregationType)