feat(eap): Mutations consumer MVP

[ayirr7]

Mutations consumer with general pattern of INSERT INTO .. SELECT FROM into Clickhouse

[untitaker]

TODO: Write cancellation rows

[ayirr7]

@untitaker 2 thoughts on this query:

1. We can rewrite this to avoid JOIN by using the WITH context. We would need to look into the performance differences.

2. What if we tried a UNION to get both the cancellation and the update in this one query? (SELECT ... UNION SELECT ...)

I'm also wondering if there's a way to do something like INSERT INTO ... VALUES (SELECT ...) (SELECT ...) so we can have 2 insertions back to back?

Just some thoughts -- I can try these out on some local data

[untitaker]

I think WITH conflicts directly with input() in my tests. I was also trying to use WITH to make the cancellation rows happen (using both WITH and UNION ALL). I'm not sure how it would help to avoid the join?

If you have a specific query let's discuss 1:1 or just push it, if it works already.

so we can have 2 insertions back to back?

then we should just merge those MutationBatches at JSON level, I don't think the query needs to adapt to accomodate larger batches.

[ayirr7]

From local testing:

• OPTIMIZE TABLE ... FINAL for some reason clears out the table. I inserted 1 "normal" row and then inserted 1 "updated" row. It seems that running optimize somehow cancelled out the two, even though they had the same sign (+1).
• Seeing 2 rows inserted instead of 1 when running snuba devserver in eap_spans_local

[untitaker]

@ayirr7 with the test command I gave you, it sends duplicate rows, and the rows contain timestamps that are immediately subjected to TTL.

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2557	1	2556	4

View the top 1 failed tests by shortest run time

tests.utils.streams.test_topics test_valid_topics

Stack Traces | 0.141s run time

Traceback (most recent call last):
  File ".../local/lib/python3.11....../site-packages/sentry_kafka_schemas/sentry_kafka_schemas.py", line 79, in get_topic
    with open(topic_path) as f:
         ^^^^^^^^^^^^^^^^
FileNotFoundError: [Errno 2] No such file or directory: '.../local/lib/python3.11.../sentry_kafka_schemas/topics/snuba-eap-mutations.yaml'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File ".../utils/streams/test_topics.py", line 10, in test_valid_topics
    sentry_kafka_schemas.get_topic(
  File ".../local/lib/python3.11....../site-packages/sentry_kafka_schemas/sentry_kafka_schemas.py", line 82, in get_topic
    raise SchemaNotFound
sentry_kafka_schemas.sentry_kafka_schemas.SchemaNotFound

To view individual test run time comparison to the main branch, go to the Test Analytics Dashboard

[ayirr7]

I have published the new topic and its schema as follows: getsentry/sentry-kafka-schemas#333. Will bump in Snuba tomorrow.

As I see it, we'll have to modify the process_message function; we'll have to do something a bit more complicated than the current

let parsed: MutationMessage = serde_json::from_slice(payload)?;

As the payload will not be directly deserialized into the MutationMessage struct. I can give this a try @untitaker ?

[ayirr7]

@untitaker I was curious why exactly we added initializing the PrimaryKey here? Is it just so that we have some consistency between the mutations and EAP spans schema?