Please address the comments from this code review:

## Individual Comments

### Comment 1
<location> `pkg/search/postgres/select_query_test.go:1032` </location>
<code_context>
 	var err error
 	switch tc.resultStruct.(type) {
 	case Struct1:
-		results, err = pgSearch.RunSelectRequestForSchema[Struct1](ctx, testDB.DB, schema.TestStructsSchema, tc.q)
+		var structs []*Struct1
+		err = pgSearch.RunSelectRequestForSchemaFn[Struct1](ctx, testDB.DB, schema.TestStructsSchema, tc.q, func(r *Struct1) error {
</code_context>

<issue_to_address>
**suggestion (testing):** Tests have been updated to use RunSelectRequestForSchemaFn, but edge cases for the callback function are not covered.

Add tests where the callback returns a non-nil error to confirm that RunSelectRequestForSchemaFn properly handles and propagates callback errors.

Suggested implementation:

```golang
	var err error
	switch tc.resultStruct.(type) {
	case Struct1:
		var structs []*Struct1
		// Normal callback test
		err = pgSearch.RunSelectRequestForSchemaFn[Struct1](ctx, testDB.DB, schema.TestStructsSchema, tc.q, func(r *Struct1) error {
			structs = append(structs, r)
			return nil
		})
		results = structs

		// Edge case: callback returns error
		expectedErr := errors.New("callback error")
		errCallback := pgSearch.RunSelectRequestForSchemaFn[Struct1](ctx, testDB.DB, schema.TestStructsSchema, tc.q, func(r *Struct1) error {
			return expectedErr
		})
		if errCallback == nil || !errors.Is(errCallback, expectedErr) {
			t.Errorf("expected callback error to be propagated, got %v", errCallback)
		}

	case Struct2:
		var structs []*Struct2
		// Normal callback test
		err = pgSearch.RunSelectRequestForSchemaFn[Struct2](ctx, testDB.DB, schema.TestStructsSchema, tc.q, func(r *Struct2) error {
			structs = append(structs, r)
			return nil
		})
		results = structs

		// Edge case: callback returns error
		expectedErr := errors.New("callback error")
		errCallback := pgSearch.RunSelectRequestForSchemaFn[Struct2](ctx, testDB.DB, schema.TestStructsSchema, tc.q, func(r *Struct2) error {
			return expectedErr
		})
		if errCallback == nil || !errors.Is(errCallback, expectedErr) {
			t.Errorf("expected callback error to be propagated, got %v", errCallback)
		}

```

- Ensure that `errors` and `testing` packages are imported at the top of the file if not already present:
  ```go
  import (
      "errors"
      "testing"
      // other imports...
  )
  ```
- If this code is inside a loop or table-driven test, you may want to use unique error messages or subtests for clarity.
</issue_to_address>

### Comment 2
<location> `pkg/search/postgres/select.go:171` </location>
<code_context>
 	return parsedQuery, nil
 }

+func retryableRunSelectOneForSchema[T any](ctx context.Context, db postgres.DB, query *query) (*T, error) {
+	if len(query.SelectedFields) == 0 {
+		return nil, errors.New("select fields required for select query")
</code_context>

<issue_to_address>
**issue (complexity):** Consider introducing a shared helper function to unify repeated query execution and error handling logic.

```suggestion
Introduce a shared helper to collapse the repeated “check-fields → Retry2 → tracedQuery → scan” boilerplate. For example:

```go
// runSelectWithRetry wraps panic recovery, retry logic, tracedQuery and scanning.
func runSelectWithRetry[T any](
    ctx context.Context,
    db postgres.DB,
    query *query,
    scanFn func(rows pgx.Rows) error,
) error {
    if len(query.SelectedFields) == 0 {
        return errors.New("select fields required for select query")
    }

    // convert panics to errors:
    var panicErr error
    defer func() {
        if r := recover(); r != nil {
            panicErr = fmt.Errorf("unexpected panic: %v", r)
        }
    }()

    return pgutils.Retry2(ctx, func() error {
        if panicErr != nil {
            return panicErr
        }
        sql := query.AsSQL()
        rows, err := tracedQuery(ctx, db, sql, query.Data...)
        if err != nil {
            return errors.Wrapf(err, "executing %s", sql)
        }
        defer rows.Close()
        return scanFn(rows)
    })
}
```

Then collapse your two helpers into thin wrappers:

```go
func retryableRunSelectOneForSchema[T any](ctx context.Context, db postgres.DB, query *query) (*T, error) {
    var result T
    err := runSelectWithRetry(ctx, db, query, func(rows pgx.Rows) error {
        return scanAPI.ScanOne(&result, rows)
    })
    return &result, pgutils.ErrNilIfNoRows(err)
}

func retryableRunSelectRequestForSchemaFn[T any](
    ctx context.Context,
    db postgres.DB,
    query *query,
    fn func(*T) error,
) error {
    return runSelectWithRetry(ctx, db, query, func(rows pgx.Rows) error {
        for rows.Next() {
            var elem T
            if err := scanAPI.Scan(&elem, rows); err != nil {
                return err
            }
            if err := fn(&elem); err != nil {
                return err
            }
        }
        return nil
    })
}
```

This removes duplicated checks, panic-to-error, retry logic, query execution and defer-Close calls, while preserving all existing behavior.
</issue_to_address>

### Comment 3
<location> `central/reports/scheduler/v2/reportgenerator/report_gen_impl.go:261` </location>
<code_context>
-			deployedImagesQueryParts.Schema, query)
+		err = pgSearch.RunSelectRequestForSchemaFn[ImageCVEQueryResponse](reportGenCtx, rg.db,
+			deployedImagesQueryParts.Schema, query, func(r *ImageCVEQueryResponse) error {
+				cveResponses = append(cveResponses, r)
+				numDeployedImageResults++
+				return nil
</code_context>

<issue_to_address>
**issue (complexity):** Consider refactoring to use a shared helper function for streaming query results and measuring lengths before and after appending, rather than maintaining per-row counters and duplicate closures.

Here’s a small refactor that preserves the streaming callback but drops your per‐row counters and duplicate closures. Instead you measure lengths before/after the callback and compute “new” rows in one go:

```go
// helper to stream into a slice
func streamImageCVEs(
    ctx context.Context,
    db DB,
    schema string,
    query search.Query,
) ([]*ImageCVEQueryResponse, error) {
    var out []*ImageCVEQueryResponse
    err := pgSearch.RunSelectRequestForSchemaFn[ImageCVEQueryResponse](
        ctx, db, schema, query,
        func(r *ImageCVEQueryResponse) error {
            out = append(out, r)
            return nil
        },
    )
    return out, err
}
```

Then in your main logic:

```go
// Deployed
start := len(cveResponses)
if filterOnImageType(..., DEPLOYED) {
    deployed, err := streamImageCVEs(reportGenCtx, rg.db,
        deployedImagesQueryParts.Schema, query)
    if err != nil {
        return nil, errors.Wrap(err, "…deployed images")
    }
    numDeployedImageResults = len(deployed)
    cveResponses = append(cveResponses, deployed...)
}

// Watched
start = len(cveResponses)
if filterOnImageType(..., WATCHED) && len(watchedImages) > 0 {
    watched, err := streamImageCVEs(reportGenCtx, rg.db,
        watchedImagesQueryParts.Schema, watchedQuery)
    if err != nil {
        return nil, errors.Wrap(err, "…watched images")
    }
    numWatchedImageResults = len(watched)
    cveResponses = append(cveResponses, watched...)
}
```

Advantages:

  * No per‐row counter in the closure
  * Shared helper for both queries
  * Clear “before/after” length measurement rather than manual increments
</issue_to_address>