Change Summary

Make consumer fetching a list of OpenLineage events instead of fetching them one-by-one:

• Add DTO property unique_key and method merge. This allows tracking multiple versions of the same DTO, and then merge into one DTO having all the non-null fields. For example, all RunDTOs with same id are merged into one RunDTO with non-null created_at and ended_at, and so on.
• Add BatchExtractResult class which tracks each type of DTO, merges old items with new ones, and then allows to iterate other merged items.
• Move all extraction logic from consumer to dedicated function extract_batch. It iterates other list of OpenLineage events (e.g. sequence of STARTED+RUNNING+SUCCESS) into a final one (SUCCESS with started_at + ended_at + ...).
• Combine all DB logic inside consumer into one function which just iterates other BatchExtractResult, and creates all the extracted items in database. All cross-DTO links are resolved by BatchExtractResult. For example, when dataset_dto.id = ... is populated from database model, all InputDTOs and OutputDTOs with the same dataset got the same dataset id, which simplifies consumer logic.

This reduces the number of IO operations a lot - instead of sequence of INSERT+UPDATE+UPDATE+UPDATE+... statements, we perform only one INSERT/UPDATE per unique object.

Benchmarks were run in the same environment (4 FastStream workers, 4 Kafka partitions) and the amount of data (239k events, 632MB with ZSTD compression).

Before this change:

• Consuming all events: 63 min, ~62 RPS.
• Postgres network usage: 611MB (input) / 764MB (output).
• Postgres IO: 30.3MB (read) / 2.81GB (write)

After this change:

• Consuming all events: 6.6 min, ~600 RPS.
• Postgres network usage: 167MB (input) / 104MB (output).
• Postgres IO: 21.3MB (read) / 1.63GB (write)

So x10 in RPS, -86% of Postgres requests, -47% of IO.

Related issue number

Checklist

• Commit message and PR title is comprehensive
• Keep the change as small as possible
• Unit and integration tests for the changes exist
• Tests pass on CI and coverage does not decrease
• Documentation reflects the changes where applicable
• docs/changelog/next_release/<pull request or issue id>.<change type>.rst file added describing change
  (see CONTRIBUTING.rst for details.)
• My PR is ready to review.