Trt 2709 jobrunid mapping csr post backfill

pkg/api/componentreadiness/dataprovider/postgres/provider.go

@@ -395,6 +395,8 @@ FROM target_tests tt
 JOIN prow_job_run_tests pjrt ON pjrt.test_id = tt.test_id
     AND (tt.suite_id = pjrt.suite_id OR (tt.suite_id IS NULL AND pjrt.suite_id IS NULL))
 JOIN prow_job_runs pjr ON pjr.id = pjrt.prow_job_run_id
+    AND pjr.prow_job_release = pjrt.prow_job_run_release
+    AND pjr.timestamp = pjrt.prow_job_run_timestamp
 JOIN prow_jobs pj ON pj.id = pjr.prow_job_id
 JOIN tests t ON t.id = pjrt.test_id
 WHERE pj.release = ?

pkg/api/jobrunscan/reevaluate.go

@@ -542,7 +542,10 @@ func (r *ReEvaluator) updatePostgresLabels(ctx context.Context, buildID string,
 			// subtraction now (idempotent -- a no-op if the row already carries
 			// the label). The label itself is written by the full-array replace
 			// below, so the merged set is used as-is.
-			if err := infrafailure.SubtractNewInfraFailure(tx, int64(jobRun.ID)); err != nil { //nolint:gosec // G115: prow_job_runs.id is a PostgreSQL serial, always within int64 range
+			if err := infrafailure.SubtractNewInfraFailure(tx, int64(jobRun.ID), query.ProwJobRunPartitionKeys{ //nolint:gosec // G115: prow_job_runs.id is a PostgreSQL serial, always within int64 range
+				ProwJobRelease: jobRun.ProwJobRelease,
+				Timestamp:      jobRun.Timestamp,
+			}); err != nil {
 				return fmt.Errorf("subtracting infra-failure summaries for build %s: %w", buildID, err)
 			}
 		} else if slices.Contains(currentRun.Labels, infrafailure.LabelInfraFailure) {

pkg/api/recent_test_failures.go

@@ -247,7 +247,9 @@ func fetchOutputs(
 	}
 
 	if err := dbc.DB.Table("prow_job_run_tests pjrt").
-		Joins("JOIN prow_job_runs pjr ON pjr.id = pjrt.prow_job_run_id").
+		Joins(`JOIN prow_job_runs pjr ON pjr.id = pjrt.prow_job_run_id
+			AND pjr.prow_job_release = pjrt.prow_job_run_release
+			AND pjr.timestamp = pjrt.prow_job_run_timestamp`).
 		Joins("JOIN prow_jobs pj ON pj.id = pjr.prow_job_id").
 		Joins(`LEFT JOIN prow_job_run_test_outputs pjrto ON pjrto.prow_job_run_test_id = pjrt.id
 			AND pjrto.prow_job_run_test_release = pjrt.prow_job_run_release

pkg/dataloader/prowloader/pgwriter/pgwriter.go

@@ -280,7 +280,6 @@ func insertJobRunIDMap(ctx context.Context, tx pgx.Tx) error {
 		INSERT INTO prow_job_run_id_map (id, prow_job_release, timestamp)
 		SELECT id, prow_job_release, timestamp
 		FROM tmp_prow_job_runs
-		ON CONFLICT (id) DO NOTHING
 	`); err != nil {
 		return fmt.Errorf("inserting prow_job_run_id_map: %w", err)
 	}

pkg/dataloader/prowloader/prow.go

@@ -545,8 +545,8 @@ func (pl *ProwLoader) findNewJobRunIDs(ctx context.Context, candidateIDs []uint)
 
 	rows, err := conn.Query(ctx, `
 		SELECT t.id FROM tmp_candidate_ids t
-		LEFT JOIN prow_job_runs r ON r.id = t.id
-		WHERE r.id IS NULL
+		LEFT JOIN prow_job_run_id_map m ON m.id = t.id
+		WHERE m.id IS NULL
 	`)
 	if err != nil {
 		return nil, fmt.Errorf("querying new job run IDs: %w", err)

pkg/db/functions.go

@@ -86,6 +86,8 @@ WITH retests AS (
         SELECT prow_jobs.id, COUNT(*) as cnt
         FROM prow_job_runs
              INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id
+                 AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release
+                 AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp
              INNER JOIN prow_pull_requests on prow_pull_requests.id = prow_job_run_prow_pull_requests.prow_pull_request_id
              INNER JOIN prow_jobs ON prow_job_runs.prow_job_id = prow_jobs.id
         WHERE prow_pull_requests.merged_at BETWEEN p_start AND p_endstamp
@@ -168,6 +170,8 @@ FROM results
              SELECT prow_pull_requests.org, prow_pull_requests.repo, prow_jobs.id
              FROM prow_job_runs
                   INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id
+                 AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release
+                 AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp
                   INNER JOIN prow_pull_requests on prow_pull_requests.id = prow_job_run_prow_pull_requests.prow_pull_request_id
                   INNER JOIN prow_jobs ON prow_job_runs.prow_job_id = prow_jobs.id
              WHERE prow_job_runs.prow_job_release = p_release

pkg/db/infrafailure/infrafailure.go

@@ -13,6 +13,7 @@ package infrafailure
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	log "github.com/sirupsen/logrus"
@@ -81,6 +82,8 @@ const setInfraFailureLabelSQL = `
 UPDATE prow_job_runs
 SET labels = array_append(labels, 'InfraFailure')
 WHERE id = ?
+  AND prow_job_release = ?
+  AND timestamp = ?
   AND (labels IS NULL OR NOT (labels @> ARRAY['InfraFailure']))`
 
 // createDeltasTempTableSQL materializes one job run's test results into a temp
@@ -192,10 +195,19 @@ func RecordInfraFailureWithOutcome(ctx context.Context, dbc *gorm.DB, prowJobRun
 func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, error) {
 	logger := log.WithField("prowJobRunID", prowJobRunID)
 
+	partKeys, err := query.LookupProwJobRunPartitionKeys(tx, prowJobRunID)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			logger.Debug("prow job run not found in id map; nothing to label")
+			return OutcomeRunNotFound, nil
+		}
+		return OutcomeUnknown, fmt.Errorf("reading partition keys for prow_job_run %d: %w", prowJobRunID, err)
+	}
+
 	// Conditional UPDATE as the first operation: set the label and acquire the
 	// row lock together. RowsAffected == 0 means either the label was already set
 	// (subtraction already done) or the run does not exist in PostgreSQL.
-	res := tx.Exec(setInfraFailureLabelSQL, prowJobRunID)
+	res := tx.Exec(setInfraFailureLabelSQL, prowJobRunID, partKeys.ProwJobRelease, partKeys.Timestamp)
 	if res.Error != nil {
 		return OutcomeUnknown, fmt.Errorf("setting InfraFailure label on prow_job_run %d: %w", prowJobRunID, res.Error)
 	}
@@ -204,7 +216,9 @@ func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, err
 		// check (cheap, and the row lock the UPDATE would have taken is moot here)
 		// so callers can distinguish an already-labeled run from a missing one.
 		var exists int
-		check := tx.Raw("SELECT 1 FROM prow_job_runs WHERE id = ? LIMIT 1", prowJobRunID).Scan(&exists)
+		check := tx.Raw(
+			"SELECT 1 FROM prow_job_runs WHERE id = ? AND prow_job_release = ? AND timestamp = ? LIMIT 1",
+			prowJobRunID, partKeys.ProwJobRelease, partKeys.Timestamp).Scan(&exists)
 		if check.Error != nil {
 			return OutcomeUnknown, fmt.Errorf("checking existence of prow_job_run %d: %w", prowJobRunID, check.Error)
 		}
@@ -219,7 +233,7 @@ func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, err
 
 	// The atomic gate passed (the label was newly applied), so remove the run's
 	// contribution from the summary tables in the same transaction.
-	if err := subtractFromSummaries(tx, prowJobRunID); err != nil {
+	if err := subtractFromSummaries(tx, prowJobRunID, partKeys); err != nil {
 		return OutcomeUnknown, err
 	}
 	return OutcomeSubtracted, nil
@@ -231,17 +245,9 @@ func recordInfraFailureInTx(tx *gorm.DB, prowJobRunID int64) (RecordOutcome, err
 // own the label and any gating that decides whether the subtraction should run.
 // All work happens on the supplied transaction so it commits or rolls back
 // atomically with the caller's other writes.
-func subtractFromSummaries(tx *gorm.DB, prowJobRunID int64) error {
+func subtractFromSummaries(tx *gorm.DB, prowJobRunID int64, partKeys query.ProwJobRunPartitionKeys) error {
 	logger := log.WithField("prowJobRunID", prowJobRunID)
 
-	// Read the run's partition keys (release and timestamp) so the delta scan
-	// below can prune to the run's single partition instead of scanning every
-	// partition for the run id.
-	partKeys, err := query.LookupProwJobRunPartitionKeys(tx, prowJobRunID)
-	if err != nil {
-		return fmt.Errorf("reading partition keys for prow_job_run %d: %w", prowJobRunID, err)
-	}
-
 	// Drop any stale temp table before recreating it. ON COMMIT DROP ties the
 	// table's lifetime to the outermost transaction, not to a savepoint, so when
 	// the subtraction runs twice inside the same outer transaction (tx is
@@ -291,17 +297,17 @@ func subtractFromSummaries(tx *gorm.DB, prowJobRunID int64) error {
 // Callers must run this inside the same row-locked transaction that later
 // replaces the labels array so the containment check and the subtraction stay
 // consistent with a concurrent RecordInfraFailure.
-func SubtractNewInfraFailure(tx *gorm.DB, prowJobRunID int64) error {
+func SubtractNewInfraFailure(tx *gorm.DB, prowJobRunID int64, partKeys query.ProwJobRunPartitionKeys) error {
 	var found int
 	res := tx.Raw(
-		"SELECT 1 FROM prow_job_runs WHERE id = ? AND labels @> ARRAY[?] LIMIT 1",
-		prowJobRunID, LabelInfraFailure).Scan(&found)
+		"SELECT 1 FROM prow_job_runs WHERE id = ? AND prow_job_release = ? AND timestamp = ? AND labels @> ARRAY[?] LIMIT 1",
+		prowJobRunID, partKeys.ProwJobRelease, partKeys.Timestamp, LabelInfraFailure).Scan(&found)
 	if res.Error != nil {
 		return fmt.Errorf("checking InfraFailure label on prow_job_run %d: %w", prowJobRunID, res.Error)
 	}
 	if res.RowsAffected > 0 {
 		log.WithField("prowJobRunID", prowJobRunID).Debug("prow job run already labeled InfraFailure; summary subtraction already applied, skipping")
 		return nil
 	}
-	return subtractFromSummaries(tx, prowJobRunID)
+	return subtractFromSummaries(tx, prowJobRunID, partKeys)
 }

pkg/db/query/job_queries.go

@@ -15,19 +15,19 @@ import (
 
 // ProwJobRunPartitionKeys holds the partition key columns for prow_job_runs.
 // Used for two-step lookups: fetch these lightweight keys first, then load the
-// full row with partition pruning. This will be replaced by a mapping table in
-// a future iteration.
+// full row with partition pruning.
 type ProwJobRunPartitionKeys struct {
 	ProwJobRelease string    `gorm:"column:prow_job_release"`
 	Timestamp      time.Time `gorm:"column:timestamp"`
 }
 
 // LookupProwJobRunPartitionKeys fetches the partition keys for a prow_job_run
-// by ID. This is intended as the first step of a two-step lookup pattern where
-// the caller then uses these keys to load the full row with partition pruning.
+// by ID from prow_job_run_id_map. This is the first step of a two-step lookup
+// pattern where the caller then uses these keys to load the full row with
+// partition pruning.
 func LookupProwJobRunPartitionKeys(gormDB *gorm.DB, jobRunID int64) (ProwJobRunPartitionKeys, error) {
 	var keys ProwJobRunPartitionKeys
-	err := gormDB.Table("prow_job_runs").
+	err := gormDB.Model(&models.ProwJobRunIDMap{}).
 		Select("prow_job_release, timestamp").
 		Where("id = ?", jobRunID).
 		Take(&keys).Error

pkg/db/query/pull_request_queries.go

@@ -26,7 +26,7 @@ func PullRequestReport(dbc *db.DB, filterOpts *filter.FilterOptions, release str
 		Joins("LEFT JOIN (?) nightly ON nightly.url = prow_pull_requests.link",
 			dbc.DB.Table("(?) as nightly", firstPayloadsByStreamAndArch).Where("nightly.stream = 'nightly' AND nightly.architecture = 'amd64'")).
 		Joins("INNER JOIN prow_job_run_prow_pull_requests ON prow_job_run_prow_pull_requests.prow_pull_request_id = prow_pull_requests.id").
-		Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id").
+		Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp").
 		Joins("INNER JOIN prow_jobs on prow_job_runs.prow_job_id = prow_jobs.id").
 		Where("prow_jobs.release = ?", release).
 		Where("prow_job_runs.prow_job_release = ?", release).
@@ -49,7 +49,7 @@ func PullRequestReport(dbc *db.DB, filterOpts *filter.FilterOptions, release str
 func PullRequestAveragePremergeFailures(dbc *db.DB, release string, start, end *time.Time) *gorm.DB {
 	premergeFailures := dbc.DB.Table("prow_job_runs").
 		Select("prow_jobs.id as prow_job_id, prow_jobs.name as prow_job_name, prow_pull_requests.org, prow_pull_requests.repo, prow_pull_requests.link, COUNT(*) as total_runs").
-		Joins("INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release").
+		Joins("INNER JOIN prow_job_run_prow_pull_requests on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp").
 		Joins("INNER JOIN prow_pull_requests on prow_pull_requests.id = prow_job_run_prow_pull_requests.prow_pull_request_id").
 		Joins("INNER JOIN prow_jobs ON prow_job_runs.prow_job_id = prow_jobs.id").
 		Where("prow_job_runs.prow_job_release = ?", release).

pkg/db/query/repository_queries.go

@@ -21,7 +21,7 @@ func RepositoryReport(dbc *db.DB, filterOpts *filter.FilterOptions, release stri
 
 	repos := dbc.DB.Table("prow_pull_requests").
 		Joins("INNER JOIN prow_job_run_prow_pull_requests ON prow_job_run_prow_pull_requests.prow_pull_request_id = prow_pull_requests.id").
-		Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id").
+		Joins("INNER JOIN prow_job_runs on prow_job_run_prow_pull_requests.prow_job_run_id = prow_job_runs.id AND prow_job_run_prow_pull_requests.prow_job_run_release = prow_job_runs.prow_job_release AND prow_job_run_prow_pull_requests.prow_job_run_timestamp = prow_job_runs.timestamp").
 		Joins("INNER JOIN prow_jobs on prow_job_runs.prow_job_id = prow_jobs.id").
 		Joins("LEFT JOIN (?) revert_count ON revert_count.org = prow_pull_requests.org AND revert_count.repo = prow_pull_requests.repo", revertCount).
 		Joins("LEFT JOIN (?) premerge_failures ON premerge_failures.prow_job_ID = prow_jobs.id", averageByJob).

pkg/sippyserver/server.go

@@ -1036,7 +1036,7 @@ func (s *Server) jsonBackendDisruptionByRun(w http.ResponseWriter, req *http.Req
 
 	var minTime, maxTime time.Time
 	if s.db != nil {
-		row := s.db.DB.Raw("SELECT MIN(timestamp), MAX(timestamp) FROM prow_job_runs WHERE id IN ?", jobRunNames).Row()
+		row := s.db.DB.Raw("SELECT MIN(timestamp), MAX(timestamp) FROM prow_job_run_id_map WHERE id IN ?", jobRunNames).Row()
 		if err := row.Scan(&minTime, &maxTime); err != nil {
 			log.WithError(err).Warn("could not look up job run timestamps from postgres, falling back to no time bound")
 		}

test/integration/job_run_id_map_test.go

@@ -0,0 +1,29 @@
+package integration
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+
+	v1 "github.com/openshift/sippy/pkg/apis/sippyprocessing/v1"
+	"github.com/openshift/sippy/pkg/db/query"
+	intutil "github.com/openshift/sippy/test/integration/util"
+)
+
+func TestLookupProwJobRunPartitionKeys(t *testing.T) {
+	dbc := intutil.NewTestDB(t, pgContainer)
+	job := intutil.CreateProwJob(t, dbc, "periodic-e2e-aws", "4.18", nil)
+	ts := time.Date(2026, 7, 15, 10, 0, 0, 0, time.UTC)
+	run := intutil.CreateProwJobRun(t, dbc, job.ID, "4.18", ts, true, v1.JobSucceeded)
+
+	keys, err := query.LookupProwJobRunPartitionKeys(dbc.DB, int64(run.ID))
+	require.NoError(t, err)
+	assert.Equal(t, "4.18", keys.ProwJobRelease)
+	assert.True(t, ts.Equal(keys.Timestamp))
+
+	_, err = query.LookupProwJobRunPartitionKeys(dbc.DB, 999999)
+	require.ErrorIs(t, err, gorm.ErrRecordNotFound)
+}