feat(cdc) Actually load the csv file and import snapshot#407
Conversation
|
|
||
| class GroupedMessageProcessor(CdcProcessor): | ||
| @dataclass(frozen=True) | ||
| class GroupedMessageRow(CDCMessageRow): |
There was a problem hiding this comment.
Maybe we will be able to move the clickhouse schema here and make it mostly automatically generated.
| with open(table_path, "rb") as table_file: | ||
| yield table_file | ||
| with open(table_path, "r") as table_file: | ||
| csv_file = csv.DictReader(table_file) |
There was a problem hiding this comment.
I am still playing with csv reader parameters to be able to load null values and escape value properly.
|
|
||
| class HTTPBatchWriter(BatchWriter): | ||
| def __init__(self, schema, host, port, options=None): | ||
| def __init__(self, schema, host, port, options=None, table_name=None): |
There was a problem hiding this comment.
It feels a little clunky that this overrides the schema attribute but I don't have any constructive ideas without giving it more thought.
There was a problem hiding this comment.
I think the only alternative way is to fit the temp_table into the dataset structure. Which defies the purpose of allowing the caller of the load script (ops generally) to handle the temporary table on their own.
| ).to_clickhouse() | ||
|
|
||
| def _process_delete(self, offset, key): | ||
| pass |
There was a problem hiding this comment.
Why is this different than the other _process_* family of messages that are now generalized?
There was a problem hiding this comment.
There was one but I cannot remember. Let me try to do it again, and then it should come back.
There was a problem hiding this comment.
oh yes. It was because, while for updates, we can simply create a new GroupedMessageRow with the new value since that value includes the identity key, for deletion this is more complex because we have to create a GroupedMessageRow with only the identity key populated. The rest is None and it is not in the WAL message.
As a result we still need a method somewhere that knows which fields to populate as identity key (only id here) and what's their name on the GroupedMessageRow class. So I could put that logic in GroupedMessageRow adding a new "forDeletion" method but that does not change much with respect to what we do here.
tkaemming
left a comment
tkaemming
left a comment
There was a problem hiding this comment.
A few small things, mostly type nits and a couple style things inline. Nothing structural or major, should be ok after those are fixed. Really the only blocker is there is an incorrect type definition in snuba/writer.py.
| last_seen: datetime | ||
| first_seen: datetime | ||
| active_at: Optional[datetime] = None | ||
| first_release_id: Optional[int] = None |
|
|
||
| def __build_record(self, offset, columnnames, columnvalues): | ||
| @dataclass(frozen=True) | ||
| class GroupedMessageRow(CDCMessageRow): |
There was a problem hiding this comment.
This wrapper around the record data could probably be made into a generic later but probably best not to worry about that until there is a use case for it
There was a problem hiding this comment.
The issue is the identity key. Just the fact that soon we will add project to the GroupedMessage makes it not generic anymore. Unless we make the parent class a "Generic" with identity as parameter. Which is something we can consider.
| offset: int, | ||
| columnnames: Sequence[str], | ||
| columnvalues: Sequence[Any], | ||
| ) -> Optional[WriterTableRow]: |
There was a problem hiding this comment.
This is helpful, I didn't realize these are optional. Why does this differ than return type of CdcMessageRow.to_clickhouse()? Shouldn't they be equivalent?
There was a problem hiding this comment.
They differ because I wanted to let the dataset specific class decide whether to ignore a WAL row. For example we could have a dataset without deletions. In that case WAL deletion should produce no effect and just return None. Essentially we let the dataset a bit more autonomy on how to interpret the WAL if they want to override these methods.
On the other hand, when we have a valid CdcMessageRow instantiated, that has to be serialized into a write to clickhouse. I cannot think about a legit case where we would be able to instantiate CdcMessageRow but we may decide to return None
## Bump sentry-conventions to 0.11.0 Updates `sentry-conventions` from 0.10.0 to [0.11.0](https://github.com/getsentry/sentry-conventions/releases/tag/0.11.0). ### Changes in 0.11.0 **New Features** - Add `session.id` attribute (#412) - Update several attributes to latest OTel representation (#418) - Add `db.operation.batch.size` (#407) - Add `app.vitals.start.prewarmed` attribute (#379) - Add and deprecate runtime context attributes (#383) - Add Android Runtime (ART) GC and memory attributes (#382) - Add `db.stored_procedure.name` and ensure span name attribute consistency (#398) - Add Cloudflare SDK attributes (#392) - Add missing GCP and FaaS attributes (#403) **Bug Fixes** - Set `pii: 'maybe'` on `faas` string attributes (#415) - Add missing Cloudflare visibility attributes (#408) **Internal** - Remove `sdks` field from attribute schema and definitions (#410)
2 participants
This builds on #398 and adds the logic that actually processes the csv table files and fills the corresponding Clickhouse table.
Specifically:
Limitation:
It does not yet support all the csv formatting that postgres can add, which I will support separately This is not relevant yet for groupedmessage where nothing is nullable and there are no strings.