Incremental Sync

[Lorenzo (llorenzinho)]

Hey everyone. I'm struggling to develop a connector with incremental sync using the ui builder.
As an example i've developed a simple rest api which expose a single endpoint returning a bunch of mocked data. Here an example:

[
  {
    "id": 0,
    "name": "Darlene",
    "surname": "Hall",
    "timestamp": "2024-07-26T10:55:04"
  },
  {
    "id": 1,
    "name": "William",
    "surname": "Martinez",
    "timestamp": "2024-07-26T23:28:26"
  }
]

I need to "incremental sync" based on timestamp. Here the connector yaml:

version: 3.8.2

type: DeclarativeSource

check:
  type: CheckStream
  stream_names:
    - data

definitions:
  streams:
    data:
      type: DeclarativeStream
      name: data
      primary_key:
        - id
      retriever:
        type: SimpleRetriever
        requester:
          $ref: '#/definitions/base_requester'
          path: data
          http_method: GET
        record_selector:
          type: RecordSelector
          extractor:
            type: DpathExtractor
            field_path: []
      incremental_sync:
        type: DatetimeBasedCursor
        cursor_field: timestamp
        cursor_datetime_formats:
          - %Y-%m-%dT%H:%M:%S
        datetime_format: '%Y-%m-%dT%H:%M:%S'
        start_datetime:
          type: MinMaxDatetime
          datetime: '{{ config["start_date"] }}'
          datetime_format: '%Y-%m-%dT%H:%M:%SZ'
        start_time_option:
          type: RequestOption
          inject_into: request_parameter
          field_name: filter
      schema_loader:
        type: InlineSchemaLoader
        schema:
          $ref: '#/schemas/data'
  base_requester:
    type: HttpRequester
    url_base: http://mock_server:8000/api

streams:
  - $ref: '#/definitions/streams/data'

spec:
  type: Spec
  connection_specification:
    type: object
    $schema: http://json-schema.org/draft-07/schema#
    required:
      - start_date
    properties:
      start_date:
        type: string
        title: Start date
        format: date-time
        pattern: ^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$
        order: 0
    additionalProperties: true

metadata:
  autoImportSchema:
    data: true

schemas:
  data:
    type: object
    $schema: http://json-schema.org/schema#
    additionalProperties: true
    properties:
      id:
        type: number
      name:
        type:
          - string
          - 'null'
      surname:
        type:
          - string
          - 'null'
      timestamp:
        type: string
    required:
      - id
      - timestamp

Here the config for the Incremental sync:

The problem here is that the state seems to be not updated. In fact, when i test the connector, the api returns data correctly, but the connector state seems to be empy:

[
  {
    "type": "STREAM",
    "stream": {
      "stream_descriptor": {
        "name": "Data"
      },
      "stream_state": {}
    },
    "sourceStats": {
      "recordCount": 10
    }
  }
]

Anyone who can help me?

[Octopyth (octopyth)]

I experienced similar symptoms in a slightly different scenario - with stream partitions. Check this article:
https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/incremental-syncs

My issue was due to different datetime format between cursor_datetime_formats and datetime_format (the format sent to the API). I got the issue resolved by creating a column that contains the partition.
transformations:
- type: AddFields
fields:
- type: AddedFieldDefinition
path:
- stream_partition
value: "{{ stream_partition.start_time }}"

With you the formats are the same, however I think it would still help if you add end_datetime, step and cursor_granularity.
end_datetime:
type: MinMaxDatetime
datetime: "{{ now_utc() }}"
datetime_format: "%Y-%m-%dT%H:%M:%S"
step: P1D
cursor_granularity: P1D

Hope this helps.