TRT-2856: Extend partition start bound to cover outlier job start times

pkg/dataloader/prowloader/prow.go

@@ -181,22 +181,40 @@ func (pl *ProwLoader) Errors() []error {
 	return pl.errors
 }
 
+// partitionStartDate computes the start of the date range for which partitions must
+// exist to accommodate the given prowJobs. loadSince is used as a floor, with a 1 day
+// grace period since bq imports based on modified time which can include job_run_start_time
+// a day earlier (see https://github.com/openshift/sippy/blob/main/pkg/dataloader/prowloader/prow.go#L473).
+//
+// That grace period covers the common case, but some jobs (e.g. ones Prow eventually marks
+// as aborted after getting stuck) can report a StartTime days before their completion time,
+// which is what the BQ query actually filters on. So the start bound is extended to the
+// earliest job we're actually about to write, to guarantee a partition exists for every row.
+func partitionStartDate(loadSince time.Time, prowJobs []prow.ProwJob) time.Time {
+	startDate := loadSince.AddDate(0, 0, -1)
+	for i := range prowJobs {
+		if st := prowJobs[i].Status.StartTime; !st.IsZero() && st.Before(startDate) {
+			startDate = st
+		}
+	}
+	return startDate
+}
+
 // ensurePartitions creates necessary partitions for partitioned tables.
 // It uses the release list from pl.releases and determines the date range based on:
-//   - pl.loadSince if available, otherwise looks back one week
+//   - pl.loadSince if available, otherwise looks back one week, plus a 1 day grace period
+//   - the earliest prowJobs StartTime, in case it falls outside the above window
 //   - Creates partitions 2 days forward from now
-func (pl *ProwLoader) ensurePartitions() error {
-	startDate := pl.resolveLoadSince()
+func (pl *ProwLoader) ensurePartitions(prowJobs []prow.ProwJob) error {
+	startDate := partitionStartDate(pl.resolveLoadSince(), prowJobs)
 
 	// Create partitions 2 days forward from now
 	endDate := time.Now().AddDate(0, 0, 2)
 
 	log.Infof("Ensuring partitions for releases %v from %s to %s",
 		pl.releases, startDate.Format("2006-01-02"), endDate.Format("2006-01-02"))
 
-	// https://github.com/openshift/sippy/blob/main/pkg/dataloader/prowloader/prow.go#L473 bq imports based on modified time which can include job_run_start_time a day earlier
-	// add grace to ensure we have a valid partition for new dbs.
-	count, err := pl.dbc.EnsurePartitions(pl.releases, startDate.AddDate(0, 0, -1), endDate, false)
+	count, err := pl.dbc.EnsurePartitions(pl.releases, startDate, endDate, false)
 	if err != nil {
 		return fmt.Errorf("failed to ensure partitions: %w", err)
 	}
@@ -239,7 +257,7 @@ func (pl *ProwLoader) Load() {
 	}
 
 	// Ensure we have partitions for the new data
-	if err := pl.ensurePartitions(); err != nil {
+	if err := pl.ensurePartitions(prowJobs); err != nil {
 		pl.errors = append(pl.errors, errors.Wrap(err, "failed to ensure partitions"))
 		return
 	}

pkg/dataloader/prowloader/prow_test.go

@@ -312,3 +312,55 @@ func TestMatchRelease(t *testing.T) {
 		})
 	}
 }
+
+func TestPartitionStartDate(t *testing.T) {
+	loadSince := time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC)
+	graceAdjusted := loadSince.AddDate(0, 0, -1)
+
+	prowJobWithStart := func(start time.Time) prow.ProwJob {
+		return prow.ProwJob{Status: prow.ProwJobStatus{StartTime: start}}
+	}
+
+	tests := []struct {
+		name     string
+		prowJobs []prow.ProwJob
+		expected time.Time
+	}{
+		{
+			name:     "no jobs falls back to loadSince with grace period",
+			prowJobs: nil,
+			expected: graceAdjusted,
+		},
+		{
+			name: "all jobs within the grace window leave the bound unchanged",
+			prowJobs: []prow.ProwJob{
+				prowJobWithStart(graceAdjusted.AddDate(0, 0, 1)),
+				prowJobWithStart(graceAdjusted),
+			},
+			expected: graceAdjusted,
+		},
+		{
+			name: "zero-value start times are ignored",
+			prowJobs: []prow.ProwJob{
+				{},
+				prowJobWithStart(graceAdjusted),
+			},
+			expected: graceAdjusted,
+		},
+		{
+			name: "outlier job start time extends the bound",
+			prowJobs: []prow.ProwJob{
+				prowJobWithStart(graceAdjusted),
+				// e.g. a job Prow marked aborted days after it actually started
+				prowJobWithStart(loadSince.AddDate(0, 0, -5)),
+			},
+			expected: loadSince.AddDate(0, 0, -5),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.True(t, tt.expected.Equal(partitionStartDate(loadSince, tt.prowJobs)))
+		})
+	}
+}