Enable incremental sync for gitlab jobs stream

[patygold]

Currently, the GitLab jobs stream is only available for full-refresh sync.
This is problematic, as it's taking hours to finish the sync.

Even though the Jobs endpoint in GitLab doesn't accept a date filter, the parent endpoint (pipelines) does, and jobs is already a child stream in the current implementation of the GitLab connector.

I propose activating the incremental_dependency in the jobs stream, and using the pipelines filter.

The code in the yml file for the jobs stream would look like this:

  jobs_stream:
    name: "jobs"
    $ref: "#/definitions/base_projects_incremental_child_stream"
    retriever:
      type: SimpleRetriever
      record_selector: "#/definitions/selector"
      paginator: "#/definitions/paginator"
      requester:
        $ref: '#/definitions/requester'
        path: 'projects/{{ stream_slice.parent_slice.id }}/pipelines/{{ stream_slice.pipeline_id }}/jobs'
      partition_router:
        type: SubstreamPartitionRouter
        parent_stream_configs:
          - type: ParentStreamConfig
            stream: "#/definitions/pipelines_stream"
            parent_key: id
            partition_field: pipeline_id
            parent_stream_name: pipelines
            incremental_dependency: true
    transformations:
      - type: AddFields
        fields:
          - type: AddedFieldDefinition
            path: ["project_id"]
            value: "{{ stream_slice.parent_slice.id }}"
          - type: AddedFieldDefinition
            path: ["user_id"]
            value: "{{ (record.get('user') or {}).get('id') }}"
          - type: AddedFieldDefinition
            path: ["pipeline_id"]
            value: "{{ (record.get('pipeline') or {}).get('id') }}"
          - type: AddedFieldDefinition
            path: ["runner_id"]
            value: "{{ (record.get('runner') or {}).get('id') }}"
          - type: AddedFieldDefinition
            path: ["commit_id"]
            value: "{{ (record.get('commit') or {}).get('id') }}"
          - type: AddedFieldDefinition
            path:
              - updated_at
            value: "{{ (record.get('pipeline') or {}).get('updated_at') }}"
    $parameters:
      path: "projects/{{ stream_slice.parent_slice.id }}/pipelines/{{ stream_slice.id }}/jobs"

And we also would need to add the field updated_at to the jobs.json file.

I've tested this on Airbyte 0.50.30 and 1.3.1, both works. What I don't know, if this will work for bigger companies, as the state file can be quite big, since it saves the state for each pipeline id.

Eg, this is what the state file can look like for the stream jobs with a couple of pipelines in the same project.

  {
    "streamDescriptor": {
      "name": "jobs"
    },
    "streamState": {
      "state": {
        "updated_at": "2024-12-19T15:07:27.208Z"
      },
      "states": [
        {
          "cursor": {
            "updated_at": "2024-12-19T14:29:59.491Z"
          },
          "partition": {
            "pipeline_id": 1596203888,
            "parent_slice": {
              "id": 45849616,
              "parent_slice": {}
            }
          }
        },
        {
          "cursor": {
            "updated_at": "2024-12-19T14:34:45.121Z"
          },
          "partition": {
            "pipeline_id": 1596203623,
            "parent_slice": {
              "id": 45849616,
              "parent_slice": {}
            }
          }
        }
     ]
   }
}