Optimize UnnestedInFilters rewriter to use an index-only scan when the primary key is present

What does this MR do and why?

Describe in detail what your merge request does and why.

Fixes: #377484 (closed)

If the primary key is available, then we will further optimize the UnnestedInFilters query rewriting by using only indexed columns for the initial filtering, allowing us to perform an index-only scan. Then, select the remaining columns using the primary key.

Before

https://console.postgres.ai/shared/9ca82f55-6a09-4b6a-9215-16cf7e8019ce

SELECT 
  "vulnerability_reads".* 
FROM 
  unnest(
    ARRAY(
      SELECT 
        "namespaces"."id" 
      FROM 
        "namespaces" 
      WHERE 
        "namespaces"."type" = 'Group' 
        AND (
          traversal_ids @> ('{9970}' :: integer[])
        )
    ) :: bigint[]
  ) AS "namespace_ids"("namespace_id"), 
  unnest(
    '{6,2,5,3,1,0,4,99}' :: smallint[]
  ) AS "report_types"("report_type"), 
  unnest('{1,4}' :: smallint[]) AS "states"("state"), 
  LATERAL (
    SELECT 
      "vulnerability_reads".* 
    FROM 
      "vulnerability_reads" 
    WHERE 
      (
        vulnerability_reads."namespace_id" = "namespace_ids"."namespace_id"
      ) 
      AND (
        vulnerability_reads."report_type" = "report_types"."report_type"
      ) 
      AND (
        vulnerability_reads."state" = "states"."state"
      ) 
    ORDER BY 
      "vulnerability_reads"."severity" DESC, 
      "vulnerability_reads"."vulnerability_id" DESC 
    LIMIT 
      101
  ) AS vulnerability_reads 
ORDER BY 
  "vulnerability_reads"."severity" DESC, 
  "vulnerability_reads"."vulnerability_id" DESC 
LIMIT 
  101;

After

https://console.postgres.ai/shared/7ef96972-bd19-4916-92c8-1c2221bdf6ec

SELECT
    "vulnerability_reads".*
FROM
    "vulnerability_reads"
WHERE
    "vulnerability_reads"."vulnerability_id" IN (
        SELECT
            "vulnerability_reads"."vulnerability_id"
        FROM
            unnest(
                ARRAY(
                    SELECT
                        "namespaces"."id"
                    FROM
                        "namespaces"
                    WHERE
                        "namespaces"."type" = 'Group'
                        AND (traversal_ids @> ('{9970}'))
                ) :: bigint []
            ) AS "namespace_ids"("namespace_id"),
            unnest('{6,2,5,3,1,0,4,99}' :: smallint []) AS "report_types"("report_type"),
            unnest('{1,4}' :: smallint []) AS "states"("state"),
            LATERAL (
                SELECT
                    "vulnerability_reads"."namespace_id",
                    "vulnerability_reads"."report_type",
                    "vulnerability_reads"."state",
                    "vulnerability_reads"."severity",
                    "vulnerability_reads"."vulnerability_id"
                FROM
                    "vulnerability_reads"
                WHERE
                    (
                        vulnerability_reads."namespace_id" = "namespace_ids"."namespace_id"
                    )
                    AND (
                        vulnerability_reads."report_type" = "report_types"."report_type"
                    )
                    AND (vulnerability_reads."state" = "states"."state")
                ORDER BY
                    "vulnerability_reads"."severity" DESC,
                    "vulnerability_reads"."vulnerability_id" DESC
                LIMIT
                    101
            ) AS vulnerability_reads
        ORDER BY
            "vulnerability_reads"."severity" DESC,
            "vulnerability_reads"."vulnerability_id" DESC
        LIMIT
            101
    )
ORDER BY
    "vulnerability_reads"."severity" DESC,
    "vulnerability_reads"."vulnerability_id" DESC
LIMIT
    101;

How to set up and validate locally

Numbered steps to set up and validate the change are strongly suggested.

1. Ensure your GDK is Ultimate licensed

2. Start the rails console: bundle exec rails c

3. Paste this ruby code:

   Click to expand

   query = <<~EOQ
   fragment VulnerabilityLocation on VulnerabilityLocation {
     ... on VulnerabilityLocationClusterImageScanning {
       image
       kubernetesResource {
         agent {
           id
           name
           webPath
         }
       }
     }
     ... on VulnerabilityLocationContainerScanning {
       image
     }
     ... on VulnerabilityLocationDependencyScanning {
       blobPath
       file
     }
     ... on VulnerabilityLocationSast {
       blobPath
       file
       startLine
     }
     ... on VulnerabilityLocationSecretDetection {
       blobPath
       file
       startLine
     }
     ... on VulnerabilityLocationDast {
       path
     }
   }
   
   fragment VulnerabilityFragment on Vulnerability {
     id
     title
     state
     severity
     detectedAt
     vulnerabilityPath
     resolvedOnDefaultBranch
     userNotesCount
     falsePositive @include(if: $vetEnabled)
     issueLinks {
       nodes {
         id
         issue {
           id
           iid
           webUrl
           webPath
           title
           state
         }
       }
     }
     identifiers {
       externalType
       name
     }
     location {
       ...VulnerabilityLocation
     }
     project {
       id
       nameWithNamespace
     }
     reportType
     scanner {
       id
       vendor
     }
   }
   
   fragment PageInfo on PageInfo {
     hasNextPage
     hasPreviousPage
     startCursor
     endCursor
   }
   
   
   query groupVulnerabilities(
     $fullPath: ID!
     $before: String
     $after: String
     $first: Int = 20
     $last: Int
     $projectId: [ID!]
     $severity: [VulnerabilitySeverity!]
     $reportType: [VulnerabilityReportType!]
     $scanner: [String!]
     $scannerId: [VulnerabilitiesScannerID!]
     $state: [VulnerabilityState!]
     $sort: VulnerabilitySort
     $hasIssues: Boolean
     $hasResolution: Boolean
     $vetEnabled: Boolean = false
     $clusterAgentId: [ClustersAgentID!]
   ) {
     group(fullPath: $fullPath) {
       id
       vulnerabilities(
         before: $before
         after: $after
         first: $first
         last: $last
         severity: $severity
         reportType: $reportType
         scanner: $scanner
         scannerId: $scannerId
         state: $state
         projectId: $projectId
         sort: $sort
         hasIssues: $hasIssues
         hasResolution: $hasResolution
         clusterAgentId: $clusterAgentId
       ) {
         nodes {
           ...VulnerabilityFragment
         }
         pageInfo {
           ...PageInfo
         }
       }
     }
   }
   EOQ
   
   variables = JSON.parse('{
     "first": 100,
     "vetEnabled": true,
     "fullPath": "gitlab-org",
     "sort": "severity_desc",
     "includeExternalIssueLinks": false,
     "last": null,
     "state": [
       "DETECTED",
       "CONFIRMED"
     ],
     "severity": [],
     "reportType": [
       "API_FUZZING",
       "CONTAINER_SCANNING",
       "COVERAGE_FUZZING",
       "DAST",
       "DEPENDENCY_SCANNING",
       "SAST",
       "SECRET_DETECTION",
       "GENERIC"
     ],
     "projectId": []
   }'.delete("\n"))
   
   current_user = User.find(1)
   
   GitlabSchema.execute(query, variables: variables, context: { current_user: current_user })

4. Check the query used for Vulnerability::Read load

MR acceptance checklist

This checklist encourages us to confirm any changes have been analyzed to reduce risks in quality, performance, reliability, security, and maintainability.

• I have evaluated the MR acceptance checklist for this MR.