From 331899d15e6d6f6acfaf7488b99dd4f4f48c30f2 Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Fri, 9 Aug 2019 15:00:24 -0700 Subject: [PATCH 01/12] Actually import snapshot into clickhouse table --- snuba/datasets/cdc/groupedmessage.py | 1 + .../datasets/cdc/groupedmessage_processor.py | 97 +++++++++++++------ snuba/snapshots/__init__.py | 24 ++++- snuba/snapshots/bulk_load.py | 12 ++- snuba/snapshots/postgres_snapshot.py | 44 ++++++++- 5 files changed, 142 insertions(+), 36 deletions(-) diff --git a/snuba/datasets/cdc/groupedmessage.py b/snuba/datasets/cdc/groupedmessage.py index a10cb1f6da2..45ad72760a7 100644 --- a/snuba/datasets/cdc/groupedmessage.py +++ b/snuba/datasets/cdc/groupedmessage.py @@ -52,6 +52,7 @@ def __init__(self): def get_bulk_loader(self, source, dest_table): return SingleTableBulkLoader( + writer=self.get_writer(), source=source, source_table=self.POSTGRES_TABLE, dest_table=dest_table, diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index 7262ad04136..d21ca58f8a1 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -1,7 +1,62 @@ +from datetime import datetime +from typing import Any, Mapping, Optional + +from dataclasses import dataclass from dateutil.parser import parse as dateutil_parse from snuba.datasets.cdc.cdcprocessors import CdcProcessor +@dataclass(frozen=True) +class GroupedMessageRow: + offset: int + id: int + record_deleted: bool + status: Optional[int] + last_seen: Optional[datetime] + first_seen: Optional[datetime] + active_at: Optional[datetime] + first_release_id: Optional[int] + + @classmethod + def from_wal(cls, offset, columnnames, columnvalues): + raw_data = dict(zip(columnnames, columnvalues)) + return GroupedMessageRow( + offset=offset, + id=raw_data['id'], + record_deleted=False, + status=raw_data['status'], + last_seen=dateutil_parse(raw_data['last_seen']), + first_seen=dateutil_parse(raw_data['first_seen']), + active_at=dateutil_parse(raw_data['active_at']), + first_release_id=raw_data['first_release_id'], + ) + + @classmethod + def from_bulk(cls, row): + return GroupedMessageRow( + offset=0, + id=int(row['id']), + record_deleted=False, + status=int(row['status']), + last_seen=dateutil_parse(row['last_seen']), + first_seen=dateutil_parse(row['first_seen']), + active_at=dateutil_parse(row['active_at']), + first_release_id=int(row['first_release_id']) if row['first_release_id'] else None, + ) + + def to_clickhouse(self) -> Mapping[str, Any]: + return { + 'offset': self.offset, + 'id': self.id, + 'record_deleted': 1 if self.record_deleted else 0, + 'status': self.status, + 'last_seen': self.last_seen, + 'first_seen': self.first_seen, + 'active_at': self.active_at, + 'first_release_id': self.first_release_id, + } + + class GroupedMessageProcessor(CdcProcessor): def __init__(self, postgres_table): @@ -9,38 +64,22 @@ def __init__(self, postgres_table): pg_table=postgres_table, ) - def __build_record(self, offset, columnnames, columnvalues): - raw_data = dict(zip(columnnames, columnvalues)) - output = { - 'offset': offset, - 'id': raw_data['id'], - 'record_deleted': 0, - 'status': raw_data['status'], - 'last_seen': dateutil_parse(raw_data['last_seen']), - 'first_seen': dateutil_parse(raw_data['first_seen']), - 'active_at': dateutil_parse(raw_data['active_at']), - 'first_release_id': raw_data['first_release_id'], - } - - return output - def _process_insert(self, offset, columnnames, columnvalues): - return self.__build_record(offset, columnnames, columnvalues) + return GroupedMessageRow.from_wal( + offset, + columnnames, + columnvalues + ).to_clickhouse() def _process_delete(self, offset, key): key_names = key['keynames'] key_values = key['keyvalues'] id = key_values[key_names.index('id')] - return { - 'offset': offset, - 'id': id, - 'record_deleted': 1, - 'status': None, - 'last_seen': None, - 'first_seen': None, - 'active_at': None, - 'first_release_id': None, - } + return GroupedMessageRow( + offset=offset, + id=id, + record_deleted=True, + ).to_clickhouse() def _process_update(self, offset, key, columnnames, columnvalues): new_id = columnvalues[columnnames.index('id')] @@ -51,4 +90,8 @@ def _process_update(self, offset, key, columnnames, columnvalues): # clickhouse will use the identity column to find rows to merge. # if we change it, merging won't work. assert old_id == new_id, 'Changing Primary Key is not supported.' - return self.__build_record(offset, columnnames, columnvalues) + return GroupedMessageRow.from_wal( + offset, + columnnames, + columnvalues + ).to_clickhouse() diff --git a/snuba/snapshots/__init__.py b/snuba/snapshots/__init__.py index af8a761672e..e3d9c0bc727 100644 --- a/snuba/snapshots/__init__.py +++ b/snuba/snapshots/__init__.py @@ -1,9 +1,10 @@ from __future__ import annotations from abc import ABC, abstractmethod +from collections.abc import Iterator from contextlib import contextmanager -from typing import NewType, Generator, IO, Optional, Sequence +from typing import Any, Mapping, NewType, Generator, IO, Optional, Sequence from dataclasses import dataclass SnapshotId = NewType("SnapshotId", str) @@ -26,6 +27,27 @@ class SnapshotDescriptor: id: SnapshotId tables: Sequence[TableConfig] + def get_table(self, table_name: str): + for t in self.tables: + if t.table == table_name: + return t + return None + + +class Table(Iterator): + + @abstractmethod + def get_name(self) -> str: + raise NotImplementedError + + @abstractmethod + def __iter__(self) -> Table: + raise NotImplementedError + + @abstractmethod + def __next__(self) -> Mapping[str, Any]: + raise NotImplementedError + class BulkLoadSource(ABC): """ diff --git a/snuba/snapshots/bulk_load.py b/snuba/snapshots/bulk_load.py index 41ef578ec26..344c83b65dd 100644 --- a/snuba/snapshots/bulk_load.py +++ b/snuba/snapshots/bulk_load.py @@ -24,13 +24,15 @@ class SingleTableBulkLoader(BulkLoader): """ def __init__(self, + writer, source: BulkLoadSource, dest_table: str, - dataset_table: str, + source_table: str, ): self.__source = source self.__dest_table = dest_table self.__source_table = source_table + self.__writer = writer def load(self) -> None: logger = logging.getLogger('snuba.bulk-loader') @@ -50,6 +52,8 @@ def load(self) -> None: logger.info("Loading snapshot %s", descriptor.id) with self.__source.get_table_file(self.__source_table) as table: - logger.info("Loading table from file %s", table.name) - # TODO: Do something with the table file - raise NotImplementedError + logger.info("Loading table from file %s", table.get_name()) + for row in table: + from snuba.datasets.cdc.groupedmessage_processor import GroupedMessageRow + clickhouse_row = GroupedMessageRow.from_bulk(row).to_clickhouse() + self.__writer.write([clickhouse_row]) diff --git a/snuba/snapshots/postgres_snapshot.py b/snuba/snapshots/postgres_snapshot.py index 08ec4566c79..151cecee1b8 100644 --- a/snuba/snapshots/postgres_snapshot.py +++ b/snuba/snapshots/postgres_snapshot.py @@ -1,5 +1,6 @@ from __future__ import annotations +import csv import jsonschema # type: ignore import json import logging @@ -7,10 +8,10 @@ from contextlib import contextmanager from dataclasses import dataclass -from typing import NewType, Generator, IO, Sequence +from typing import Any, Mapping, NewType, Generator, IO, Sequence from snuba.snapshots import SnapshotDescriptor, TableConfig -from snuba.snapshots import BulkLoadSource +from snuba.snapshots import BulkLoadSource, Table Xid = NewType("Xid", int) @@ -72,6 +73,22 @@ class PostgresSnapshotDescriptor(SnapshotDescriptor): logger = logging.getLogger('snuba.postgres-snapshot') +class PostgresCSVTableDump(Table): + + def __init__(self, csv_reader, table_name: str) -> None: + self.__csv_reader = csv_reader + self.__table_name = table_name + + def get_name(self) -> str: + return self.__table_name + + def __iter__(self) -> Table: + return self + + def __next__(self) -> Mapping[str, Any]: + return next(self.__csv_reader) + + class PostgresSnapshot(BulkLoadSource): """ TODO: Make this a library to be reused outside of Snuba when after this @@ -124,8 +141,27 @@ def get_descriptor(self) -> PostgresSnapshotDescriptor: def get_table_file(self, table: str) -> Generator[IO[bytes], None, None]: table_path = os.path.join(self.__path, "tables", "%s.csv" % table) try: - 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) + columns = csv_file.fieldnames + + expected_columns = self.__descriptor.get_table(table).columns + if expected_columns: + expected_set = set(expected_columns) + existing_set = set(columns) + + if not expected_set <= existing_set: + raise ValueError( + "The table file is missing columns %r " % expected_set - existing_set) + + if len(existing_set) != len(expected_set): + logger.warning( + "The table file contains more columns than expected %r", + existing_set - expected_set, + ) + + yield PostgresCSVTableDump(csv_file, table_file.name) + except FileNotFoundError: raise ValueError( "The snapshot does not contain the requested table %s" % table, From 07ef2db44ddd6c0aacdb6288181b20742b57746b Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Fri, 9 Aug 2019 17:40:16 -0700 Subject: [PATCH 02/12] Batch writes --- snuba/cli/bulk_load.py | 5 +++-- snuba/datasets/__init__.py | 3 ++- snuba/datasets/cdc/groupedmessage.py | 4 ++-- snuba/settings_base.py | 1 + snuba/snapshots/bulk_load.py | 26 ++++++++++++++----------- snuba/writer.py | 29 ++++++++++++++++++++++++++-- 6 files changed, 50 insertions(+), 18 deletions(-) diff --git a/snuba/cli/bulk_load.py b/snuba/cli/bulk_load.py index 8525b6e7d96..1687c277c99 100644 --- a/snuba/cli/bulk_load.py +++ b/snuba/cli/bulk_load.py @@ -32,5 +32,6 @@ def bulk_load(dataset, dest_table, source, log_level): ) loader = dataset.get_bulk_loader(snapshot_source, dest_table) - - loader.load() + # TODO: see whether we need to pass options to the writer + writer = dataset.get_writer({}, dest_table) + loader.load(writer) diff --git a/snuba/datasets/__init__.py b/snuba/datasets/__init__.py index 6e4ff8cc3fe..f5a2deaa6ab 100644 --- a/snuba/datasets/__init__.py +++ b/snuba/datasets/__init__.py @@ -22,7 +22,7 @@ def get_schema(self): def get_processor(self): return self.__processor - def get_writer(self, options=None): + def get_writer(self, options=None, table_name=None): from snuba import settings from snuba.writer import HTTPBatchWriter @@ -31,6 +31,7 @@ def get_writer(self, options=None): settings.CLICKHOUSE_HOST, settings.CLICKHOUSE_HTTP_PORT, options, + table_name, ) def default_conditions(self): diff --git a/snuba/datasets/cdc/groupedmessage.py b/snuba/datasets/cdc/groupedmessage.py index 45ad72760a7..679a4195149 100644 --- a/snuba/datasets/cdc/groupedmessage.py +++ b/snuba/datasets/cdc/groupedmessage.py @@ -1,6 +1,6 @@ from snuba.clickhouse import ColumnSet, DateTime, Nullable, UInt from snuba.datasets import Dataset -from snuba.datasets.cdc.groupedmessage_processor import GroupedMessageProcessor +from snuba.datasets.cdc.groupedmessage_processor import GroupedMessageProcessor, GroupedMessageRow from snuba.datasets.schema import ReplacingMergeTreeSchema from snuba.snapshots.bulk_load import SingleTableBulkLoader @@ -52,8 +52,8 @@ def __init__(self): def get_bulk_loader(self, source, dest_table): return SingleTableBulkLoader( - writer=self.get_writer(), source=source, source_table=self.POSTGRES_TABLE, dest_table=dest_table, + row_processor=lambda row: GroupedMessageRow.from_bulk(row).to_clickhouse(), ) diff --git a/snuba/settings_base.py b/snuba/settings_base.py index 7eee25f6bf8..8e91decacc1 100644 --- a/snuba/settings_base.py +++ b/snuba/settings_base.py @@ -43,6 +43,7 @@ # Snuba Options SNAPSHOT_LOAD_PRODUCT = 'snuba' +BULK_CLICKHOUSE_BUFFER = 1000 # Convenience columns that evaluate to a bucketed time, the # bucketing depends on the granularity parameter. diff --git a/snuba/snapshots/bulk_load.py b/snuba/snapshots/bulk_load.py index 344c83b65dd..3f79832806a 100644 --- a/snuba/snapshots/bulk_load.py +++ b/snuba/snapshots/bulk_load.py @@ -1,8 +1,12 @@ from abc import ABC, abstractmethod +from typing import Any, Callable, Mapping + import logging from snuba.clickhouse import ClickhousePool from snuba.snapshots import BulkLoadSource +from snuba.writer import BatchWriter, BufferedBatchWriter +from snuba import settings class BulkLoader(ABC): @@ -24,17 +28,17 @@ class SingleTableBulkLoader(BulkLoader): """ def __init__(self, - writer, - source: BulkLoadSource, - dest_table: str, - source_table: str, - ): + source: BulkLoadSource, + dest_table: str, + source_table: str, + row_processor: Callable[[Mapping[str, Any]], Mapping[str, Any]], + ): self.__source = source self.__dest_table = dest_table self.__source_table = source_table - self.__writer = writer + self.__row_processor = row_processor - def load(self) -> None: + def load(self, writer: BatchWriter) -> None: logger = logging.getLogger('snuba.bulk-loader') clickhouse_ro = ClickhousePool(client_settings={ @@ -53,7 +57,7 @@ def load(self) -> None: with self.__source.get_table_file(self.__source_table) as table: logger.info("Loading table from file %s", table.get_name()) - for row in table: - from snuba.datasets.cdc.groupedmessage_processor import GroupedMessageRow - clickhouse_row = GroupedMessageRow.from_bulk(row).to_clickhouse() - self.__writer.write([clickhouse_row]) + with BufferedBatchWriter(writer, settings.BULK_CLICKHOUSE_BUFFER) as buffer_writer: + for row in table: + clickhouse_data = self.__row_processor(row) + buffer_writer.write(clickhouse_data) diff --git a/snuba/writer.py b/snuba/writer.py index 5b461ddea9f..a67e2300e89 100644 --- a/snuba/writer.py +++ b/snuba/writer.py @@ -42,10 +42,11 @@ def write(self, rows): class HTTPBatchWriter(BatchWriter): - def __init__(self, schema, host, port, options=None): + def __init__(self, schema, host, port, options=None, table_name=None): self.__schema = schema self.__base_url = 'http://{host}:{port}/'.format(host=host, port=port) self.__options = options if options is not None else {} + self.__table_name = table_name or schema.get_table_name() def __default(self, value): if isinstance(value, datetime): @@ -58,8 +59,32 @@ def __encode(self, row): def write(self, rows): parameters = self.__options.copy() - parameters['query'] = "INSERT INTO {table} FORMAT JSONEachRow".format(table=self.__schema.get_table_name()) + parameters['query'] = "INSERT INTO {table} FORMAT JSONEachRow".format(table=self.__table_name) requests.post( urljoin(self.__base_url, '?' + urlencode(parameters)), data=map(self.__encode, rows), ).raise_for_status() + + +class BufferedBatchWriter: + def __init__(self, writer: BatchWriter, buffer_size: int): + self.__writer = writer + self.__buffer_size = buffer_size + self.__buffer = [] + + def __flush(self) -> None: + logger.debug("Flushing buffer with %d elements", len(self.__buffer)) + self.__writer.write(self.__buffer) + self.__buffer = [] + + def __enter__(self): + return self + + def __exit__(self, type, value, traceback): + if self.__buffer: + self.__flush() + + def write(self, row): + self.__buffer.append(row) + if len(self.__buffer) >= self.__buffer_size: + self.__flush() From 8c1d8af0d3f9c8a3205abe7b96c4714aa8c10208 Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Fri, 9 Aug 2019 18:10:17 -0700 Subject: [PATCH 03/12] Refactor record dataclass --- snuba/datasets/cdc/cdcprocessors.py | 42 +++++++++++++++++-- .../datasets/cdc/groupedmessage_processor.py | 27 ++---------- snuba/snapshots/__init__.py | 15 ------- snuba/snapshots/bulk_load.py | 2 +- snuba/snapshots/postgres_snapshot.py | 29 ++++--------- 5 files changed, 52 insertions(+), 63 deletions(-) diff --git a/snuba/datasets/cdc/cdcprocessors.py b/snuba/datasets/cdc/cdcprocessors.py index 004501e712d..68dcee86111 100644 --- a/snuba/datasets/cdc/cdcprocessors.py +++ b/snuba/datasets/cdc/cdcprocessors.py @@ -1,12 +1,30 @@ +from abc import abstractclassmethod +from typing import Any, Mapping, Type + from snuba.processor import MessageProcessor KAFKA_ONLY_PARTITION = 0 # CDC only works with single partition topics. So partition must be 0 +class CDCMessageRow: + @abstractclassmethod + def from_wal(cls, offset, columnnames, columnvalues): + raise NotImplementedError + + @abstractclassmethod + def from_bulk(cls, row): + raise NotImplementedError + + @abstractclassmethod + def to_clickhouse(self) -> Mapping[str, Any]: + raise NotImplementedError + + class CdcProcessor(MessageProcessor): - def __init__(self, pg_table): + def __init__(self, pg_table: str, message_row_class: Type[CDCMessageRow]): self.pg_table = pg_table + self._message_row_class = message_row_class def _process_begin(self, offset): pass @@ -15,10 +33,28 @@ def _process_commit(self, offset): pass def _process_insert(self, offset, columnnames, columnvalues): - pass + return self._message_row_class.from_wal( + offset, + columnnames, + columnvalues + ).to_clickhouse() def _process_update(self, offset, key, columnnames, columnvalues): - pass + old_key = dict(zip(key['keynames'], key['keyvalues'])) + new_key = { + key: columnvalues[columnnames.index(key)] + for key + in key['keynames'] + } + # We cannot support a change in the identity of the record + # clickhouse will use the identity column to find rows to merge. + # if we change it, merging won't work. + assert old_key == new_key, 'Changing Primary Key is not supported.' + return self._message_row_class.from_wal( + offset, + columnnames, + columnvalues + ).to_clickhouse() def _process_delete(self, offset, key): pass diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index d21ca58f8a1..32ea9a8bf50 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -3,11 +3,11 @@ from dataclasses import dataclass from dateutil.parser import parse as dateutil_parse -from snuba.datasets.cdc.cdcprocessors import CdcProcessor +from snuba.datasets.cdc.cdcprocessors import CdcProcessor, CDCMessageRow @dataclass(frozen=True) -class GroupedMessageRow: +class GroupedMessageRow(CDCMessageRow): offset: int id: int record_deleted: bool @@ -62,15 +62,9 @@ class GroupedMessageProcessor(CdcProcessor): def __init__(self, postgres_table): super(GroupedMessageProcessor, self).__init__( pg_table=postgres_table, + message_row_class=GroupedMessageRow, ) - def _process_insert(self, offset, columnnames, columnvalues): - return GroupedMessageRow.from_wal( - offset, - columnnames, - columnvalues - ).to_clickhouse() - def _process_delete(self, offset, key): key_names = key['keynames'] key_values = key['keyvalues'] @@ -80,18 +74,3 @@ def _process_delete(self, offset, key): id=id, record_deleted=True, ).to_clickhouse() - - def _process_update(self, offset, key, columnnames, columnvalues): - new_id = columnvalues[columnnames.index('id')] - key_names = key['keynames'] - key_values = key['keyvalues'] - old_id = key_values[key_names.index('id')] - # We cannot support a change in the identity of the record - # clickhouse will use the identity column to find rows to merge. - # if we change it, merging won't work. - assert old_id == new_id, 'Changing Primary Key is not supported.' - return GroupedMessageRow.from_wal( - offset, - columnnames, - columnvalues - ).to_clickhouse() diff --git a/snuba/snapshots/__init__.py b/snuba/snapshots/__init__.py index e3d9c0bc727..c60997e6b99 100644 --- a/snuba/snapshots/__init__.py +++ b/snuba/snapshots/__init__.py @@ -34,21 +34,6 @@ def get_table(self, table_name: str): return None -class Table(Iterator): - - @abstractmethod - def get_name(self) -> str: - raise NotImplementedError - - @abstractmethod - def __iter__(self) -> Table: - raise NotImplementedError - - @abstractmethod - def __next__(self) -> Mapping[str, Any]: - raise NotImplementedError - - class BulkLoadSource(ABC): """ Represent a source we can bulk load Snuba datasets from. diff --git a/snuba/snapshots/bulk_load.py b/snuba/snapshots/bulk_load.py index 3f79832806a..22ff44d6961 100644 --- a/snuba/snapshots/bulk_load.py +++ b/snuba/snapshots/bulk_load.py @@ -56,7 +56,7 @@ def load(self, writer: BatchWriter) -> None: logger.info("Loading snapshot %s", descriptor.id) with self.__source.get_table_file(self.__source_table) as table: - logger.info("Loading table from file %s", table.get_name()) + logger.info("Loading table from file %s", self.__source_table) with BufferedBatchWriter(writer, settings.BULK_CLICKHOUSE_BUFFER) as buffer_writer: for row in table: clickhouse_data = self.__row_processor(row) diff --git a/snuba/snapshots/postgres_snapshot.py b/snuba/snapshots/postgres_snapshot.py index 151cecee1b8..2f50d014b82 100644 --- a/snuba/snapshots/postgres_snapshot.py +++ b/snuba/snapshots/postgres_snapshot.py @@ -8,10 +8,10 @@ from contextlib import contextmanager from dataclasses import dataclass -from typing import Any, Mapping, NewType, Generator, IO, Sequence +from typing import Any, Mapping, NewType, Generator, IO, Iterable, Sequence from snuba.snapshots import SnapshotDescriptor, TableConfig -from snuba.snapshots import BulkLoadSource, Table +from snuba.snapshots import BulkLoadSource Xid = NewType("Xid", int) @@ -73,22 +73,6 @@ class PostgresSnapshotDescriptor(SnapshotDescriptor): logger = logging.getLogger('snuba.postgres-snapshot') -class PostgresCSVTableDump(Table): - - def __init__(self, csv_reader, table_name: str) -> None: - self.__csv_reader = csv_reader - self.__table_name = table_name - - def get_name(self) -> str: - return self.__table_name - - def __iter__(self) -> Table: - return self - - def __next__(self) -> Mapping[str, Any]: - return next(self.__csv_reader) - - class PostgresSnapshot(BulkLoadSource): """ TODO: Make this a library to be reused outside of Snuba when after this @@ -137,8 +121,12 @@ def load(cls, product: str, path: str) -> PostgresSnapshot: def get_descriptor(self) -> PostgresSnapshotDescriptor: return self.__descriptor + def __table_iterable(self, csv_table) -> Generator[Mapping[str, Any], None, None]: + for r in csv_table: + yield r + @contextmanager - def get_table_file(self, table: str) -> Generator[IO[bytes], None, None]: + def get_table_file(self, table: str) -> Generator[Iterable[Mapping[str, Any]], None, None]: table_path = os.path.join(self.__path, "tables", "%s.csv" % table) try: with open(table_path, "r") as table_file: @@ -160,7 +148,8 @@ def get_table_file(self, table: str) -> Generator[IO[bytes], None, None]: existing_set - expected_set, ) - yield PostgresCSVTableDump(csv_file, table_file.name) + # yield PostgresCSVTableDump(csv_file, table_file.name) + yield self.__table_iterable(csv_file) except FileNotFoundError: raise ValueError( From 02515de60317dc4d90d79c43ea902338b5e3053d Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Fri, 9 Aug 2019 18:28:57 -0700 Subject: [PATCH 04/12] Fix typing --- snuba/datasets/cdc/cdcprocessors.py | 11 +++++++++-- snuba/snapshots/__init__.py | 4 ++-- snuba/snapshots/bulk_load.py | 11 +++++++---- snuba/snapshots/postgres_snapshot.py | 10 +++++++--- snuba/writer.py | 16 +++++++++++++--- 5 files changed, 38 insertions(+), 14 deletions(-) diff --git a/snuba/datasets/cdc/cdcprocessors.py b/snuba/datasets/cdc/cdcprocessors.py index 68dcee86111..0c757507f53 100644 --- a/snuba/datasets/cdc/cdcprocessors.py +++ b/snuba/datasets/cdc/cdcprocessors.py @@ -1,4 +1,4 @@ -from abc import abstractclassmethod +from abc import ABC, abstractclassmethod from typing import Any, Mapping, Type from snuba.processor import MessageProcessor @@ -6,7 +6,14 @@ KAFKA_ONLY_PARTITION = 0 # CDC only works with single partition topics. So partition must be 0 -class CDCMessageRow: +class CDCMessageRow(ABC): + """ + Takes care of the data transformation from WAL to clickhouse and from + bulk load to clickhouse. The goal is to keep all these transformation + function in the same place because they ultimately have to be consistent + with the Clickhouse schema. + """ + @abstractclassmethod def from_wal(cls, offset, columnnames, columnvalues): raise NotImplementedError diff --git a/snuba/snapshots/__init__.py b/snuba/snapshots/__init__.py index c60997e6b99..c259836afc2 100644 --- a/snuba/snapshots/__init__.py +++ b/snuba/snapshots/__init__.py @@ -4,7 +4,7 @@ from collections.abc import Iterator from contextlib import contextmanager -from typing import Any, Mapping, NewType, Generator, IO, Optional, Sequence +from typing import Any, Mapping, NewType, Generator, IO, Iterable, Optional, Sequence from dataclasses import dataclass SnapshotId = NewType("SnapshotId", str) @@ -48,5 +48,5 @@ def get_descriptor(self) -> SnapshotDescriptor: @abstractmethod @contextmanager - def get_table_file(self, table: str) -> Generator[IO[bytes], None, None]: + def get_table_file(self, table: str) -> Generator[Iterable[Mapping[str, Any]], None, None]: raise NotImplementedError diff --git a/snuba/snapshots/bulk_load.py b/snuba/snapshots/bulk_load.py index 22ff44d6961..98e90e82ab9 100644 --- a/snuba/snapshots/bulk_load.py +++ b/snuba/snapshots/bulk_load.py @@ -5,7 +5,7 @@ from snuba.clickhouse import ClickhousePool from snuba.snapshots import BulkLoadSource -from snuba.writer import BatchWriter, BufferedBatchWriter +from snuba.writer import BatchWriter, BufferedWriterWrapper from snuba import settings @@ -18,7 +18,7 @@ class BulkLoader(ABC): the bulk load operation. """ @abstractmethod - def load(self) -> None: + def load(self, writer: BatchWriter) -> None: raise NotImplementedError @@ -56,8 +56,11 @@ def load(self, writer: BatchWriter) -> None: logger.info("Loading snapshot %s", descriptor.id) with self.__source.get_table_file(self.__source_table) as table: - logger.info("Loading table from file %s", self.__source_table) - with BufferedBatchWriter(writer, settings.BULK_CLICKHOUSE_BUFFER) as buffer_writer: + logger.info("Loading table %s from file", self.__source_table) + row_count = 0 + with BufferedWriterWrapper(writer, settings.BULK_CLICKHOUSE_BUFFER) as buffer_writer: for row in table: clickhouse_data = self.__row_processor(row) buffer_writer.write(clickhouse_data) + row_count += 1 + logger.info("Load complete %d records loaded", row_count) diff --git a/snuba/snapshots/postgres_snapshot.py b/snuba/snapshots/postgres_snapshot.py index 2f50d014b82..0e302a7cf06 100644 --- a/snuba/snapshots/postgres_snapshot.py +++ b/snuba/snapshots/postgres_snapshot.py @@ -126,7 +126,10 @@ def __table_iterable(self, csv_table) -> Generator[Mapping[str, Any], None, None yield r @contextmanager - def get_table_file(self, table: str) -> Generator[Iterable[Mapping[str, Any]], None, None]: + def get_table_file( + self, + table: str, + ) -> Generator[Iterable[Mapping[str, Any]], None, None]: table_path = os.path.join(self.__path, "tables", "%s.csv" % table) try: with open(table_path, "r") as table_file: @@ -140,15 +143,16 @@ def get_table_file(self, table: str) -> Generator[Iterable[Mapping[str, Any]], N if not expected_set <= existing_set: raise ValueError( - "The table file is missing columns %r " % expected_set - existing_set) + "The table file is missing columns %r " % (expected_set - existing_set)) if len(existing_set) != len(expected_set): logger.warning( "The table file contains more columns than expected %r", existing_set - expected_set, ) + else: + logger.info("Won't pre-validate snapshot columns. There is nothing in the descriptor") - # yield PostgresCSVTableDump(csv_file, table_file.name) yield self.__table_iterable(csv_file) except FileNotFoundError: diff --git a/snuba/writer.py b/snuba/writer.py index a67e2300e89..10927c93f00 100644 --- a/snuba/writer.py +++ b/snuba/writer.py @@ -1,6 +1,7 @@ import json import logging from datetime import datetime +from typing import Any, List, Mapping import requests from urllib.parse import urlencode, urljoin @@ -66,11 +67,20 @@ def write(self, rows): ).raise_for_status() -class BufferedBatchWriter: +class BufferedWriterWrapper: + """ + This is a wrapper that adds a buffer around a BatchWriter. + When consuming data from Kafka, the buffering logic is performed by the + batching consumer. + This is for the use cases that are not Kafka related. + + This is not thread safe. Don't try to do parallel flush hoping in the GIL. + """ + def __init__(self, writer: BatchWriter, buffer_size: int): self.__writer = writer self.__buffer_size = buffer_size - self.__buffer = [] + self.__buffer: List[Mapping[str, Any]] = [] def __flush(self) -> None: logger.debug("Flushing buffer with %d elements", len(self.__buffer)) @@ -84,7 +94,7 @@ def __exit__(self, type, value, traceback): if self.__buffer: self.__flush() - def write(self, row): + def write(self, row: Mapping[str, Any]): self.__buffer.append(row) if len(self.__buffer) >= self.__buffer_size: self.__flush() From a270659322c60028177e55039002d235aca3dc20 Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Sat, 10 Aug 2019 22:51:36 -0700 Subject: [PATCH 05/12] Fix tests --- .../datasets/cdc/groupedmessage_processor.py | 10 ++-- snuba/snapshots/postgres_snapshot.py | 12 ++-- tests/snapshots/test_postgres_snapshot.py | 57 +++++++++++++------ 3 files changed, 52 insertions(+), 27 deletions(-) diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index 32ea9a8bf50..b4305b79d7c 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -11,11 +11,11 @@ class GroupedMessageRow(CDCMessageRow): offset: int id: int record_deleted: bool - status: Optional[int] - last_seen: Optional[datetime] - first_seen: Optional[datetime] - active_at: Optional[datetime] - first_release_id: Optional[int] + status: Optional[int] = None + last_seen: Optional[datetime] = None + first_seen: Optional[datetime] = None + active_at: Optional[datetime] = None + first_release_id: Optional[int] = None @classmethod def from_wal(cls, offset, columnnames, columnvalues): diff --git a/snuba/snapshots/postgres_snapshot.py b/snuba/snapshots/postgres_snapshot.py index 0e302a7cf06..32e7a168323 100644 --- a/snuba/snapshots/postgres_snapshot.py +++ b/snuba/snapshots/postgres_snapshot.py @@ -8,7 +8,7 @@ from contextlib import contextmanager from dataclasses import dataclass -from typing import Any, Mapping, NewType, Generator, IO, Iterable, Sequence +from typing import Any, Mapping, NewType, Generator, Iterable, Sequence from snuba.snapshots import SnapshotDescriptor, TableConfig from snuba.snapshots import BulkLoadSource @@ -140,14 +140,18 @@ def get_table_file( if expected_columns: expected_set = set(expected_columns) existing_set = set(columns) - if not expected_set <= existing_set: raise ValueError( - "The table file is missing columns %r " % (expected_set - existing_set)) + "The table %s is missing columns %r " % ( + table, + expected_set - existing_set, + ) + ) if len(existing_set) != len(expected_set): logger.warning( - "The table file contains more columns than expected %r", + "The table %s contains more columns than expected %r", + table, existing_set - expected_set, ) else: diff --git a/tests/snapshots/test_postgres_snapshot.py b/tests/snapshots/test_postgres_snapshot.py index d8ec9e544bd..937a146be52 100644 --- a/tests/snapshots/test_postgres_snapshot.py +++ b/tests/snapshots/test_postgres_snapshot.py @@ -1,15 +1,9 @@ import os # NOQA +import pytest from snuba.snapshots.postgres_snapshot import PostgresSnapshot - -class TestPostgresSnapshot: - - def test_parse_snapshot(self, tmp_path): - snapshot_base = tmp_path / "cdc-snapshot" - snapshot_base.mkdir() - meta = snapshot_base / "metadata.json" - meta.write_text(""" +META_FILE = """ { "snapshot_id": "50a86ad6-b4b7-11e9-a46f-acde48001122", "product": "snuba", @@ -35,21 +29,34 @@ def test_parse_snapshot(self, tmp_path): ], "start_timestamp": 1564703503.682226 } - """) +""" + + +class TestPostgresSnapshot: + + def __prepare_directory(self, tmp_path, table_content): + snapshot_base = tmp_path / "cdc-snapshot" + snapshot_base.mkdir() + meta = snapshot_base / "metadata.json" + meta.write_text(META_FILE) tables_dir = tmp_path / "cdc-snapshot" / "tables" tables_dir.mkdir() groupedmessage = tables_dir / "sentry_groupedmessage.csv" - groupedmessage.write_text( - """id, status -0, 1 -""" - ) + groupedmessage.write_text(table_content) groupassignee = tables_dir / "sentry_groupasignee" groupassignee.write_text( - """id, project_id + """id,project_id """ ) + return snapshot_base + def test_parse_snapshot(self, tmp_path): + snapshot_base = self.__prepare_directory( + tmp_path, + """id,status +0,1 +""" + ) snapshot = PostgresSnapshot.load("snuba", snapshot_base) descriptor = snapshot.get_descriptor() assert descriptor.id == "50a86ad6-b4b7-11e9-a46f-acde48001122" @@ -65,6 +72,20 @@ def test_parse_snapshot(self, tmp_path): assert tables["sentry_groupasignee"] == ["id", "project_id"] with snapshot.get_table_file("sentry_groupedmessage") as table: - content = table.readlines() - assert content[0] == b"id, status\n" - assert content[1] == b"0, 1\n" + line = next(table) + assert line == { + "id": "0", + "status": "1", + } + + def test_parse_invalid_snapshot(self, tmp_path): + snapshot_base = self.__prepare_directory( + tmp_path, + """id +0 +""" + ) + with pytest.raises(ValueError, match=".+sentry_groupedmessage.+status.+"): + snapshot = PostgresSnapshot.load("snuba", snapshot_base) + with snapshot.get_table_file("sentry_groupedmessage") as table: + next(table) From 950571375ecf5edb6c276d34c8cc82cd8cd2f7cf Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Sat, 10 Aug 2019 23:28:31 -0700 Subject: [PATCH 06/12] Beef up tests --- tests/base.py | 38 +++++++++++++++------------- tests/test_api.py | 4 +-- tests/test_cleanup.py | 10 ++++---- tests/test_groupedmessage.py | 49 +++++++++++++++++++++++++++++++++--- tests/test_optimize.py | 10 ++++---- 5 files changed, 78 insertions(+), 33 deletions(-) diff --git a/tests/base.py b/tests/base.py index d24b9d3a2d2..f3b4d5f3cc5 100644 --- a/tests/base.py +++ b/tests/base.py @@ -30,6 +30,7 @@ def wrap_raw_event(event): 'data': event } + def get_event(): from fixtures import raw_event timestamp = datetime.utcnow() @@ -141,7 +142,24 @@ def teardown_method(self, test_method): redis_client.flushdb() -class BaseEventsTest(BaseTest): +class BaseDatasetTest(BaseTest): + def write_processed_records(self, records): + if not isinstance(records, (list, tuple)): + records = [records] + + rows = [] + for event in records: + rows.append(event) + + return self.write_rows(rows) + + def write_rows(self, rows): + if not isinstance(rows, (list, tuple)): + rows = [rows] + self.dataset.get_writer().write(rows) + + +class BaseEventsTest(BaseDatasetTest): def setup_method(self, test_method): super(BaseEventsTest, self).setup_method(test_method, 'events') self.event = get_event() @@ -168,20 +186,4 @@ def write_raw_events(self, events): _, processed = self.dataset.get_processor().process_message(event) out.append(processed) - return self.write_processed_events(out) - - def write_processed_events(self, events): - if not isinstance(events, (list, tuple)): - events = [events] - - rows = [] - for event in events: - rows.append(event) - - return self.write_rows(rows) - - def write_rows(self, rows): - if not isinstance(rows, (list, tuple)): - rows = [rows] - - self.dataset.get_writer().write(rows) + return self.write_processed_records(out) diff --git a/tests/test_api.py b/tests/test_api.py index e07def54756..792851320b6 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -102,7 +102,7 @@ def generate_fizzbuzz_events(self): } } })) - self.write_processed_events(events) + self.write_processed_records(events) def test_count(self): """ @@ -709,7 +709,7 @@ def test_doesnt_select_deletions(self): } result1 = json.loads(self.app.post('/query', data=json.dumps(query)).data) - self.write_processed_events([{ + self.write_processed_records([{ 'event_id': '9' * 32, 'project_id': 1, 'group_id': 1, diff --git a/tests/test_cleanup.py b/tests/test_cleanup.py index e552190f240..303df5611a9 100644 --- a/tests/test_cleanup.py +++ b/tests/test_cleanup.py @@ -16,7 +16,7 @@ def to_monday(d): assert parts == [] # base, 90 retention - self.write_processed_events(self.create_event_for_date(base)) + self.write_processed_records(self.create_event_for_date(base)) parts = cleanup.get_active_partitions(self.clickhouse, self.database, self.table) assert parts == [(to_monday(base), 90)] stale = cleanup.filter_stale_partitions(parts, as_of=base) @@ -24,7 +24,7 @@ def to_monday(d): # -40 days, 90 retention three_weeks_ago = base - timedelta(days=7 * 3) - self.write_processed_events(self.create_event_for_date(three_weeks_ago)) + self.write_processed_records(self.create_event_for_date(three_weeks_ago)) parts = cleanup.get_active_partitions(self.clickhouse, self.database, self.table) assert parts == [(to_monday(three_weeks_ago), 90), (to_monday(base), 90)] stale = cleanup.filter_stale_partitions(parts, as_of=base) @@ -32,7 +32,7 @@ def to_monday(d): # -100 days, 90 retention thirteen_weeks_ago = base - timedelta(days=7 * 13) - self.write_processed_events(self.create_event_for_date(thirteen_weeks_ago)) + self.write_processed_records(self.create_event_for_date(thirteen_weeks_ago)) parts = cleanup.get_active_partitions(self.clickhouse, self.database, self.table) assert parts == [ (to_monday(thirteen_weeks_ago), 90), @@ -44,7 +44,7 @@ def to_monday(d): # -1 week, 30 retention one_week_ago = base - timedelta(days=7) - self.write_processed_events(self.create_event_for_date(one_week_ago, 30)) + self.write_processed_records(self.create_event_for_date(one_week_ago, 30)) parts = cleanup.get_active_partitions(self.clickhouse, self.database, self.table) assert parts == [ (to_monday(thirteen_weeks_ago), 90), @@ -57,7 +57,7 @@ def to_monday(d): # -5 weeks, 30 retention five_weeks_ago = base - timedelta(days=7 * 5) - self.write_processed_events(self.create_event_for_date(five_weeks_ago, 30)) + self.write_processed_records(self.create_event_for_date(five_weeks_ago, 30)) parts = cleanup.get_active_partitions(self.clickhouse, self.database, self.table) assert parts == [ (to_monday(thirteen_weeks_ago), 90), diff --git a/tests/test_groupedmessage.py b/tests/test_groupedmessage.py index fa5a7e79e96..08088dfc606 100644 --- a/tests/test_groupedmessage.py +++ b/tests/test_groupedmessage.py @@ -2,12 +2,19 @@ import simplejson as json from datetime import datetime -from base import BaseTest +from base import BaseDatasetTest +from snuba.clickhouse import ClickhousePool from snuba.consumer import KafkaMessageMetadata -from snuba.datasets.cdc.groupedmessage_processor import GroupedMessageProcessor +from snuba.datasets.cdc.groupedmessage_processor import GroupedMessageProcessor, GroupedMessageRow -class TestGroupedMessageProcessor(BaseTest): +class TestGroupedMessage(BaseDatasetTest): + + def setup_method(self, test_method): + super(TestGroupedMessage, self).setup_method( + test_method, + 'groupedmessage', + ) BEGIN_MSG = '{"event":"begin","xid":2380836}' COMMIT_MSG = '{"event":"commit"}' @@ -94,6 +101,19 @@ def test_messages(self): insert_msg = json.loads(self.INSERT_MSG) ret = processor.process_message(insert_msg, metadata) assert ret[1] == self.PROCESSED + self.write_processed_records(ret[1]) + cp = ClickhousePool() + ret = cp.execute("SELECT * FROM test_groupedmessage_local;") + assert ret[0] == ( + 42, # offset + 0, # deleted + 74, # id + 0, # status + datetime(2019, 6, 19, 6, 46, 28), + datetime(2019, 6, 19, 6, 45, 32), + datetime(2019, 6, 19, 6, 45, 32), + None, + ) update_msg = json.loads(self.UPDATE_MSG) ret = processor.process_message(update_msg, metadata) @@ -102,3 +122,26 @@ def test_messages(self): delete_msg = json.loads(self.DELETE_MSG) ret = processor.process_message(delete_msg, metadata) assert ret[1] == self.DELETED + + def test_bulk_load(self): + row = GroupedMessageRow.from_bulk({ + 'id': '10', + 'status': '0', + 'last_seen': '2019-06-28 17:57:32+00', + 'first_seen': '2019-06-28 06:40:17+00', + 'active_at': '2019-06-28 06:40:17+00', + 'first_release_id': '26', + }) + self.write_processed_records(row.to_clickhouse()) + cp = ClickhousePool() + ret = cp.execute("SELECT * FROM test_groupedmessage_local;") + assert ret[0] == ( + 0, # offset + 0, # deleted + 10, # id + 0, # status + datetime(2019, 6, 28, 17, 57, 32), + datetime(2019, 6, 28, 6, 40, 17), + datetime(2019, 6, 28, 6, 40, 17), + 26, + ) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 59413a698a6..b91597f2c89 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -15,25 +15,25 @@ def test(self): base_monday = base - timedelta(days=base.weekday()) # 1 event, 0 unoptimized parts - self.write_processed_events(self.create_event_for_date(base)) + self.write_processed_records(self.create_event_for_date(base)) parts = optimize.get_partitions_to_optimize(self.clickhouse, self.database, self.table) assert parts == [] # 2 events in the same part, 1 unoptimized part - self.write_processed_events(self.create_event_for_date(base)) + self.write_processed_records(self.create_event_for_date(base)) parts = optimize.get_partitions_to_optimize(self.clickhouse, self.database, self.table) assert parts == [(base_monday, 90)] # 3 events in the same part, 1 unoptimized part - self.write_processed_events(self.create_event_for_date(base)) + self.write_processed_records(self.create_event_for_date(base)) parts = optimize.get_partitions_to_optimize(self.clickhouse, self.database, self.table) assert parts == [(base_monday, 90)] # 3 events in one part, 2 in another, 2 unoptimized parts a_month_earlier = base_monday - timedelta(days=31) a_month_earlier_monday = a_month_earlier - timedelta(days=a_month_earlier.weekday()) - self.write_processed_events(self.create_event_for_date(a_month_earlier_monday)) - self.write_processed_events(self.create_event_for_date(a_month_earlier_monday)) + self.write_processed_records(self.create_event_for_date(a_month_earlier_monday)) + self.write_processed_records(self.create_event_for_date(a_month_earlier_monday)) parts = optimize.get_partitions_to_optimize(self.clickhouse, self.database, self.table) assert parts == [(base_monday, 90), (a_month_earlier_monday, 90)] From 484f30df76660c1d3751e4e116c2dc674beb1602 Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Mon, 19 Aug 2019 18:49:01 -0700 Subject: [PATCH 07/12] Nits --- snuba/cli/bulk_load.py | 6 +++++- snuba/datasets/cdc/groupedmessage_processor.py | 6 +++--- snuba/snapshots/__init__.py | 5 +++-- snuba/snapshots/bulk_load.py | 8 ++++---- snuba/snapshots/postgres_snapshot.py | 10 +++------- snuba/writer.py | 10 ++++++---- 6 files changed, 24 insertions(+), 21 deletions(-) diff --git a/snuba/cli/bulk_load.py b/snuba/cli/bulk_load.py index 1687c277c99..95f45f9aa70 100644 --- a/snuba/cli/bulk_load.py +++ b/snuba/cli/bulk_load.py @@ -5,6 +5,7 @@ from snuba import settings from snuba.datasets.factory import get_dataset, DATASET_NAMES from snuba.snapshots.postgres_snapshot import PostgresSnapshot +from snuba.writer import BufferedWriterWrapper @click.command() @@ -33,5 +34,8 @@ def bulk_load(dataset, dest_table, source, log_level): loader = dataset.get_bulk_loader(snapshot_source, dest_table) # TODO: see whether we need to pass options to the writer - writer = dataset.get_writer({}, dest_table) + writer = BufferedWriterWrapper( + dataset.get_writer({}, dest_table), + settings.BULK_CLICKHOUSE_BUFFER, + ) loader.load(writer) diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index b4305b79d7c..37782c701ae 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -8,7 +8,7 @@ @dataclass(frozen=True) class GroupedMessageRow(CDCMessageRow): - offset: int + offset: Optional[int] id: int record_deleted: bool status: Optional[int] = None @@ -34,7 +34,7 @@ def from_wal(cls, offset, columnnames, columnvalues): @classmethod def from_bulk(cls, row): return GroupedMessageRow( - offset=0, + offset=None, id=int(row['id']), record_deleted=False, status=int(row['status']), @@ -46,7 +46,7 @@ def from_bulk(cls, row): def to_clickhouse(self) -> Mapping[str, Any]: return { - 'offset': self.offset, + 'offset': self.offset if self.offset is not None else 0, 'id': self.id, 'record_deleted': 1 if self.record_deleted else 0, 'status': self.status, diff --git a/snuba/snapshots/__init__.py b/snuba/snapshots/__init__.py index c259836afc2..a2404edd599 100644 --- a/snuba/snapshots/__init__.py +++ b/snuba/snapshots/__init__.py @@ -8,6 +8,7 @@ from dataclasses import dataclass SnapshotId = NewType("SnapshotId", str) +TableRow = Mapping[str, Any] @dataclass(frozen=True) @@ -31,7 +32,7 @@ def get_table(self, table_name: str): for t in self.tables: if t.table == table_name: return t - return None + raise ValueError("Table %s does not exists in the snapshot" % table_name) class BulkLoadSource(ABC): @@ -48,5 +49,5 @@ def get_descriptor(self) -> SnapshotDescriptor: @abstractmethod @contextmanager - def get_table_file(self, table: str) -> Generator[Iterable[Mapping[str, Any]], None, None]: + def get_table_file(self, table: str) -> Generator[Iterable[TableRow], None, None]: raise NotImplementedError diff --git a/snuba/snapshots/bulk_load.py b/snuba/snapshots/bulk_load.py index 98e90e82ab9..2eb5aac6a51 100644 --- a/snuba/snapshots/bulk_load.py +++ b/snuba/snapshots/bulk_load.py @@ -5,7 +5,7 @@ from snuba.clickhouse import ClickhousePool from snuba.snapshots import BulkLoadSource -from snuba.writer import BatchWriter, BufferedWriterWrapper +from snuba.writer import BufferedWriterWrapper from snuba import settings @@ -18,7 +18,7 @@ class BulkLoader(ABC): the bulk load operation. """ @abstractmethod - def load(self, writer: BatchWriter) -> None: + def load(self, writer: BufferedWriterWrapper) -> None: raise NotImplementedError @@ -38,7 +38,7 @@ def __init__(self, self.__source_table = source_table self.__row_processor = row_processor - def load(self, writer: BatchWriter) -> None: + def load(self, writer: BufferedWriterWrapper) -> None: logger = logging.getLogger('snuba.bulk-loader') clickhouse_ro = ClickhousePool(client_settings={ @@ -58,7 +58,7 @@ def load(self, writer: BatchWriter) -> None: with self.__source.get_table_file(self.__source_table) as table: logger.info("Loading table %s from file", self.__source_table) row_count = 0 - with BufferedWriterWrapper(writer, settings.BULK_CLICKHOUSE_BUFFER) as buffer_writer: + with writer as buffer_writer: for row in table: clickhouse_data = self.__row_processor(row) buffer_writer.write(clickhouse_data) diff --git a/snuba/snapshots/postgres_snapshot.py b/snuba/snapshots/postgres_snapshot.py index 32e7a168323..9ec7f981665 100644 --- a/snuba/snapshots/postgres_snapshot.py +++ b/snuba/snapshots/postgres_snapshot.py @@ -11,7 +11,7 @@ from typing import Any, Mapping, NewType, Generator, Iterable, Sequence from snuba.snapshots import SnapshotDescriptor, TableConfig -from snuba.snapshots import BulkLoadSource +from snuba.snapshots import BulkLoadSource, TableRow Xid = NewType("Xid", int) @@ -121,15 +121,11 @@ def load(cls, product: str, path: str) -> PostgresSnapshot: def get_descriptor(self) -> PostgresSnapshotDescriptor: return self.__descriptor - def __table_iterable(self, csv_table) -> Generator[Mapping[str, Any], None, None]: - for r in csv_table: - yield r - @contextmanager def get_table_file( self, table: str, - ) -> Generator[Iterable[Mapping[str, Any]], None, None]: + ) -> Generator[Iterable[TableRow], None, None]: table_path = os.path.join(self.__path, "tables", "%s.csv" % table) try: with open(table_path, "r") as table_file: @@ -157,7 +153,7 @@ def get_table_file( else: logger.info("Won't pre-validate snapshot columns. There is nothing in the descriptor") - yield self.__table_iterable(csv_file) + yield csv_file except FileNotFoundError: raise ValueError( diff --git a/snuba/writer.py b/snuba/writer.py index 10927c93f00..07b6837824f 100644 --- a/snuba/writer.py +++ b/snuba/writer.py @@ -11,6 +11,8 @@ logger = logging.getLogger('snuba.writer') +TableRow = Mapping[str, Any] + class BatchWriter(object): def __init__(self, schema): @@ -34,7 +36,7 @@ def __row_to_column_list(self, columns, row): values.append(value) return values - def write(self, rows): + def write(self, rows: TableRow): columns = self.__schema.get_columns() self.__connection.execute_robust("INSERT INTO %(table)s (%(colnames)s) VALUES" % { 'colnames': ", ".join(col.escaped for col in columns), @@ -55,7 +57,7 @@ def __default(self, value): else: raise TypeError - def __encode(self, row): + def __encode(self, row: TableRow): return json.dumps(row, default=self.__default).encode('utf-8') def write(self, rows): @@ -80,7 +82,7 @@ class BufferedWriterWrapper: def __init__(self, writer: BatchWriter, buffer_size: int): self.__writer = writer self.__buffer_size = buffer_size - self.__buffer: List[Mapping[str, Any]] = [] + self.__buffer: List[TableRow] = [] def __flush(self) -> None: logger.debug("Flushing buffer with %d elements", len(self.__buffer)) @@ -94,7 +96,7 @@ def __exit__(self, type, value, traceback): if self.__buffer: self.__flush() - def write(self, row: Mapping[str, Any]): + def write(self, row: TableRow): self.__buffer.append(row) if len(self.__buffer) >= self.__buffer_size: self.__flush() From d088cf5dac82b8c682a6f5491f35cfbb6ae911a4 Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Mon, 19 Aug 2019 19:02:59 -0700 Subject: [PATCH 08/12] Split record and row --- .../datasets/cdc/groupedmessage_processor.py | 51 +++++++++++-------- 1 file changed, 31 insertions(+), 20 deletions(-) diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index 37782c701ae..9e66fcaaa21 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -6,16 +6,21 @@ from snuba.datasets.cdc.cdcprocessors import CdcProcessor, CDCMessageRow +@dataclass(frozen=True) +class GroupMessageRecord: + status: int + last_seen: datetime + first_seen: datetime + active_at: Optional[datetime] = None + first_release_id: Optional[int] = None + + @dataclass(frozen=True) class GroupedMessageRow(CDCMessageRow): offset: Optional[int] id: int record_deleted: bool - status: Optional[int] = None - last_seen: Optional[datetime] = None - first_seen: Optional[datetime] = None - active_at: Optional[datetime] = None - first_release_id: Optional[int] = None + record_content: Optional[GroupMessageRecord] @classmethod def from_wal(cls, offset, columnnames, columnvalues): @@ -24,11 +29,13 @@ def from_wal(cls, offset, columnnames, columnvalues): offset=offset, id=raw_data['id'], record_deleted=False, - status=raw_data['status'], - last_seen=dateutil_parse(raw_data['last_seen']), - first_seen=dateutil_parse(raw_data['first_seen']), - active_at=dateutil_parse(raw_data['active_at']), - first_release_id=raw_data['first_release_id'], + record_content=GroupMessageRecord( + status=raw_data['status'], + last_seen=dateutil_parse(raw_data['last_seen']), + first_seen=dateutil_parse(raw_data['first_seen']), + active_at=dateutil_parse(raw_data['active_at']), + first_release_id=raw_data['first_release_id'], + ) ) @classmethod @@ -37,23 +44,27 @@ def from_bulk(cls, row): offset=None, id=int(row['id']), record_deleted=False, - status=int(row['status']), - last_seen=dateutil_parse(row['last_seen']), - first_seen=dateutil_parse(row['first_seen']), - active_at=dateutil_parse(row['active_at']), - first_release_id=int(row['first_release_id']) if row['first_release_id'] else None, + record_content=GroupMessageRecord( + status=int(row['status']), + last_seen=dateutil_parse(row['last_seen']), + first_seen=dateutil_parse(row['first_seen']), + active_at=dateutil_parse(row['active_at']), + first_release_id=int(row['first_release_id']) if row['first_release_id'] else None, + ) ) def to_clickhouse(self) -> Mapping[str, Any]: + deleted = self.record_content is None + record = self.record_content return { 'offset': self.offset if self.offset is not None else 0, 'id': self.id, 'record_deleted': 1 if self.record_deleted else 0, - 'status': self.status, - 'last_seen': self.last_seen, - 'first_seen': self.first_seen, - 'active_at': self.active_at, - 'first_release_id': self.first_release_id, + 'status': None if deleted else record.status, + 'last_seen': None if deleted else record.last_seen, + 'first_seen': None if deleted else record.first_seen, + 'active_at': None if deleted else record.active_at, + 'first_release_id': None if deleted else record.first_release_id, } From 46ea3203330a62cc337919ac85a01571f51ba025 Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Mon, 19 Aug 2019 19:04:43 -0700 Subject: [PATCH 09/12] fix test --- snuba/datasets/cdc/groupedmessage_processor.py | 1 + 1 file changed, 1 insertion(+) diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index 9e66fcaaa21..85fd6e2bb76 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -84,4 +84,5 @@ def _process_delete(self, offset, key): offset=offset, id=id, record_deleted=True, + record_content=None ).to_clickhouse() From 5e9a528c0647ef8706e87bbc3f38d166223e3a6e Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Wed, 21 Aug 2019 15:27:05 -0700 Subject: [PATCH 10/12] Fix nits --- snuba/datasets/cdc/cdcprocessors.py | 27 ++++++++++++------- .../datasets/cdc/groupedmessage_processor.py | 23 +++++++++++----- snuba/snapshots/__init__.py | 6 ++--- snuba/snapshots/bulk_load.py | 6 ++--- snuba/snapshots/postgres_snapshot.py | 4 +-- snuba/writer.py | 16 +++++------ 6 files changed, 50 insertions(+), 32 deletions(-) diff --git a/snuba/datasets/cdc/cdcprocessors.py b/snuba/datasets/cdc/cdcprocessors.py index 0c757507f53..980386618f5 100644 --- a/snuba/datasets/cdc/cdcprocessors.py +++ b/snuba/datasets/cdc/cdcprocessors.py @@ -1,12 +1,15 @@ +from __future__ import annotations + from abc import ABC, abstractclassmethod -from typing import Any, Mapping, Type +from typing import Any, Mapping, Optional, Sequence, Type from snuba.processor import MessageProcessor +from snuba.writer import WriterTableRow KAFKA_ONLY_PARTITION = 0 # CDC only works with single partition topics. So partition must be 0 -class CDCMessageRow(ABC): +class CdcMessageRow(ABC): """ Takes care of the data transformation from WAL to clickhouse and from bulk load to clickhouse. The goal is to keep all these transformation @@ -15,21 +18,27 @@ class CDCMessageRow(ABC): """ @abstractclassmethod - def from_wal(cls, offset, columnnames, columnvalues): + def from_wal(cls, + offset: int, + columnnames: Sequence[Any], + columnvalues: Sequence[Any], + ) -> CdcMessageRow: raise NotImplementedError @abstractclassmethod - def from_bulk(cls, row): + def from_bulk(cls, + row: Mapping[str, Any], + ) -> CdcMessageRow: raise NotImplementedError @abstractclassmethod - def to_clickhouse(self) -> Mapping[str, Any]: + def to_clickhouse(self) -> WriterTableRow: raise NotImplementedError class CdcProcessor(MessageProcessor): - def __init__(self, pg_table: str, message_row_class: Type[CDCMessageRow]): + def __init__(self, pg_table: str, message_row_class: Type[CdcMessageRow]): self.pg_table = pg_table self._message_row_class = message_row_class @@ -39,14 +48,14 @@ def _process_begin(self, offset): def _process_commit(self, offset): pass - def _process_insert(self, offset, columnnames, columnvalues): + def _process_insert(self, offset, columnnames, columnvalues) -> Optional[WriterTableRow]: return self._message_row_class.from_wal( offset, columnnames, columnvalues ).to_clickhouse() - def _process_update(self, offset, key, columnnames, columnvalues): + def _process_update(self, offset, key, columnnames, columnvalues) -> Optional[WriterTableRow]: old_key = dict(zip(key['keynames'], key['keyvalues'])) new_key = { key: columnvalues[columnnames.index(key)] @@ -63,7 +72,7 @@ def _process_update(self, offset, key, columnnames, columnvalues): columnvalues ).to_clickhouse() - def _process_delete(self, offset, key): + def _process_delete(self, offset, key) -> Optional[WriterTableRow]: pass def process_message(self, value, metadata): diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index 85fd6e2bb76..e5531176902 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -1,9 +1,12 @@ +from __future__ import annotations + from datetime import datetime -from typing import Any, Mapping, Optional +from typing import Any, Mapping, Optional, Sequence from dataclasses import dataclass from dateutil.parser import parse as dateutil_parse -from snuba.datasets.cdc.cdcprocessors import CdcProcessor, CDCMessageRow +from snuba.datasets.cdc.cdcprocessors import CdcProcessor, CdcMessageRow +from snuba.writer import WriterTableRow @dataclass(frozen=True) @@ -16,14 +19,18 @@ class GroupMessageRecord: @dataclass(frozen=True) -class GroupedMessageRow(CDCMessageRow): +class GroupedMessageRow(CdcMessageRow): offset: Optional[int] id: int record_deleted: bool record_content: Optional[GroupMessageRecord] @classmethod - def from_wal(cls, offset, columnnames, columnvalues): + def from_wal(cls, + offset: int, + columnnames: Sequence[Any], + columnvalues: Sequence[Any], + ) -> GroupedMessageRow: raw_data = dict(zip(columnnames, columnvalues)) return GroupedMessageRow( offset=offset, @@ -39,7 +46,9 @@ def from_wal(cls, offset, columnnames, columnvalues): ) @classmethod - def from_bulk(cls, row): + def from_bulk(cls, + row: Mapping[str, Any], + ) -> GroupedMessageRow: return GroupedMessageRow( offset=None, id=int(row['id']), @@ -53,7 +62,7 @@ def from_bulk(cls, row): ) ) - def to_clickhouse(self) -> Mapping[str, Any]: + def to_clickhouse(self) -> WriterTableRow: deleted = self.record_content is None record = self.record_content return { @@ -76,7 +85,7 @@ def __init__(self, postgres_table): message_row_class=GroupedMessageRow, ) - def _process_delete(self, offset, key): + def _process_delete(self, offset, key) -> Optional[WriterTableRow]: key_names = key['keynames'] key_values = key['keyvalues'] id = key_values[key_names.index('id')] diff --git a/snuba/snapshots/__init__.py b/snuba/snapshots/__init__.py index a2404edd599..9cd5b943a5e 100644 --- a/snuba/snapshots/__init__.py +++ b/snuba/snapshots/__init__.py @@ -8,7 +8,7 @@ from dataclasses import dataclass SnapshotId = NewType("SnapshotId", str) -TableRow = Mapping[str, Any] +SnapshotTableRow = Mapping[str, Any] @dataclass(frozen=True) @@ -32,7 +32,7 @@ def get_table(self, table_name: str): for t in self.tables: if t.table == table_name: return t - raise ValueError("Table %s does not exists in the snapshot" % table_name) + raise ValueError(f"Table {table_name} does not exists in the snapshot") class BulkLoadSource(ABC): @@ -49,5 +49,5 @@ def get_descriptor(self) -> SnapshotDescriptor: @abstractmethod @contextmanager - def get_table_file(self, table: str) -> Generator[Iterable[TableRow], None, None]: + def get_table_file(self, table: str) -> Generator[Iterable[SnapshotTableRow], None, None]: raise NotImplementedError diff --git a/snuba/snapshots/bulk_load.py b/snuba/snapshots/bulk_load.py index 2eb5aac6a51..d94bf57698b 100644 --- a/snuba/snapshots/bulk_load.py +++ b/snuba/snapshots/bulk_load.py @@ -5,8 +5,8 @@ from snuba.clickhouse import ClickhousePool from snuba.snapshots import BulkLoadSource -from snuba.writer import BufferedWriterWrapper -from snuba import settings +from snuba.writer import BufferedWriterWrapper, WriterTableRow +from snuba.snapshots import SnapshotTableRow class BulkLoader(ABC): @@ -31,7 +31,7 @@ def __init__(self, source: BulkLoadSource, dest_table: str, source_table: str, - row_processor: Callable[[Mapping[str, Any]], Mapping[str, Any]], + row_processor: Callable[[SnapshotTableRow], WriterTableRow], ): self.__source = source self.__dest_table = dest_table diff --git a/snuba/snapshots/postgres_snapshot.py b/snuba/snapshots/postgres_snapshot.py index 9ec7f981665..71688ccf889 100644 --- a/snuba/snapshots/postgres_snapshot.py +++ b/snuba/snapshots/postgres_snapshot.py @@ -11,7 +11,7 @@ from typing import Any, Mapping, NewType, Generator, Iterable, Sequence from snuba.snapshots import SnapshotDescriptor, TableConfig -from snuba.snapshots import BulkLoadSource, TableRow +from snuba.snapshots import BulkLoadSource, SnapshotTableRow Xid = NewType("Xid", int) @@ -125,7 +125,7 @@ def get_descriptor(self) -> PostgresSnapshotDescriptor: def get_table_file( self, table: str, - ) -> Generator[Iterable[TableRow], None, None]: + ) -> Generator[Iterable[SnapshotTableRow], None, None]: table_path = os.path.join(self.__path, "tables", "%s.csv" % table) try: with open(table_path, "r") as table_file: diff --git a/snuba/writer.py b/snuba/writer.py index 685bb73a9ad..f0308b4fd92 100644 --- a/snuba/writer.py +++ b/snuba/writer.py @@ -1,7 +1,7 @@ import json import logging from datetime import datetime -from typing import Any, List, Mapping +from typing import Any, Iterable, List, Mapping from urllib3.connectionpool import HTTPConnectionPool from urllib3.exceptions import HTTPError @@ -12,7 +12,7 @@ logger = logging.getLogger('snuba.writer') -TableRow = Mapping[str, Any] +WriterTableRow = Mapping[str, Any] class BatchWriter(object): @@ -37,7 +37,7 @@ def __row_to_column_list(self, columns, row): values.append(value) return values - def write(self, rows: TableRow): + def write(self, rows: Iterable[WriterTableRow]): columns = self.__schema.get_columns() self.__connection.execute_robust("INSERT INTO %(table)s (%(colnames)s) VALUES" % { 'colnames': ", ".join(col.escaped for col in columns), @@ -58,12 +58,12 @@ def __default(self, value): else: raise TypeError - def __encode(self, row: TableRow): + def __encode(self, row: WriterTableRow): return json.dumps(row, default=self.__default).encode('utf-8') - def write(self, rows): + def write(self, rows: Iterable[WriterTableRow]): parameters = self.__options.copy() - parameters['query'] = f"INSERT INTO {self.__schema.get_table_name()} FORMAT JSONEachRow" + parameters['query'] = f"INSERT INTO {self.__table_name} FORMAT JSONEachRow" resp = self.__pool.urlopen( 'POST', '/?' + urlencode(parameters), @@ -91,7 +91,7 @@ class BufferedWriterWrapper: def __init__(self, writer: BatchWriter, buffer_size: int): self.__writer = writer self.__buffer_size = buffer_size - self.__buffer: List[TableRow] = [] + self.__buffer: List[WriterTableRow] = [] def __flush(self) -> None: logger.debug("Flushing buffer with %d elements", len(self.__buffer)) @@ -105,7 +105,7 @@ def __exit__(self, type, value, traceback): if self.__buffer: self.__flush() - def write(self, row: TableRow): + def write(self, row: WriterTableRow): self.__buffer.append(row) if len(self.__buffer) >= self.__buffer_size: self.__flush() From 2ad23c87c4c1b23fa527cfa1f72675ddd8c460fa Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Wed, 21 Aug 2019 15:33:34 -0700 Subject: [PATCH 11/12] Some additional typing --- snuba/datasets/cdc/cdcprocessors.py | 22 ++++++++++++++----- .../datasets/cdc/groupedmessage_processor.py | 5 ++++- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/snuba/datasets/cdc/cdcprocessors.py b/snuba/datasets/cdc/cdcprocessors.py index 980386618f5..e38a47a2e29 100644 --- a/snuba/datasets/cdc/cdcprocessors.py +++ b/snuba/datasets/cdc/cdcprocessors.py @@ -42,20 +42,29 @@ def __init__(self, pg_table: str, message_row_class: Type[CdcMessageRow]): self.pg_table = pg_table self._message_row_class = message_row_class - def _process_begin(self, offset): + def _process_begin(self, offset: int): pass - def _process_commit(self, offset): + def _process_commit(self, offset: int): pass - def _process_insert(self, offset, columnnames, columnvalues) -> Optional[WriterTableRow]: + def _process_insert(self, + offset: int, + columnnames: Sequence[str], + columnvalues: Sequence[Any], + ) -> Optional[WriterTableRow]: return self._message_row_class.from_wal( offset, columnnames, columnvalues ).to_clickhouse() - def _process_update(self, offset, key, columnnames, columnvalues) -> Optional[WriterTableRow]: + def _process_update(self, + offset: int, + key: Mapping[str, Any], + columnnames: Sequence[str], + columnvalues: Sequence[Any], + ) -> Optional[WriterTableRow]: old_key = dict(zip(key['keynames'], key['keyvalues'])) new_key = { key: columnvalues[columnnames.index(key)] @@ -72,7 +81,10 @@ def _process_update(self, offset, key, columnnames, columnvalues) -> Optional[Wr columnvalues ).to_clickhouse() - def _process_delete(self, offset, key) -> Optional[WriterTableRow]: + def _process_delete(self, + offset: int, + key: Mapping[str, Any], + ) -> Optional[WriterTableRow]: pass def process_message(self, value, metadata): diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index e5531176902..af14a8f9ae4 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -85,7 +85,10 @@ def __init__(self, postgres_table): message_row_class=GroupedMessageRow, ) - def _process_delete(self, offset, key) -> Optional[WriterTableRow]: + def _process_delete(self, + offset: int, + key: Mapping[str, Any], + ) -> Optional[WriterTableRow]: key_names = key['keynames'] key_values = key['keyvalues'] id = key_values[key_names.index('id')] From 05e29660e32b1c09a216bd1cf46b62b13502eaee Mon Sep 17 00:00:00 2001 From: Filippo Pacifici Date: Wed, 21 Aug 2019 15:52:36 -0700 Subject: [PATCH 12/12] Fix some types --- snuba/datasets/cdc/cdcprocessors.py | 16 +++++++++------- snuba/datasets/cdc/groupedmessage_processor.py | 13 ++++++------- 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/snuba/datasets/cdc/cdcprocessors.py b/snuba/datasets/cdc/cdcprocessors.py index e38a47a2e29..76476ea2452 100644 --- a/snuba/datasets/cdc/cdcprocessors.py +++ b/snuba/datasets/cdc/cdcprocessors.py @@ -1,6 +1,6 @@ from __future__ import annotations -from abc import ABC, abstractclassmethod +from abc import ABC, abstractmethod from typing import Any, Mapping, Optional, Sequence, Type from snuba.processor import MessageProcessor @@ -17,21 +17,23 @@ class CdcMessageRow(ABC): with the Clickhouse schema. """ - @abstractclassmethod - def from_wal(cls, + @classmethod + def from_wal( + cls, offset: int, - columnnames: Sequence[Any], + columnnames: Sequence[str], columnvalues: Sequence[Any], ) -> CdcMessageRow: raise NotImplementedError - @abstractclassmethod - def from_bulk(cls, + @classmethod + def from_bulk( + cls, row: Mapping[str, Any], ) -> CdcMessageRow: raise NotImplementedError - @abstractclassmethod + @abstractmethod def to_clickhouse(self) -> WriterTableRow: raise NotImplementedError diff --git a/snuba/datasets/cdc/groupedmessage_processor.py b/snuba/datasets/cdc/groupedmessage_processor.py index af14a8f9ae4..e1cbc4132ba 100644 --- a/snuba/datasets/cdc/groupedmessage_processor.py +++ b/snuba/datasets/cdc/groupedmessage_processor.py @@ -28,7 +28,7 @@ class GroupedMessageRow(CdcMessageRow): @classmethod def from_wal(cls, offset: int, - columnnames: Sequence[Any], + columnnames: Sequence[str], columnvalues: Sequence[Any], ) -> GroupedMessageRow: raw_data = dict(zip(columnnames, columnvalues)) @@ -63,17 +63,16 @@ def from_bulk(cls, ) def to_clickhouse(self) -> WriterTableRow: - deleted = self.record_content is None record = self.record_content return { 'offset': self.offset if self.offset is not None else 0, 'id': self.id, 'record_deleted': 1 if self.record_deleted else 0, - 'status': None if deleted else record.status, - 'last_seen': None if deleted else record.last_seen, - 'first_seen': None if deleted else record.first_seen, - 'active_at': None if deleted else record.active_at, - 'first_release_id': None if deleted else record.first_release_id, + 'status': None if not record else record.status, + 'last_seen': None if not record else record.last_seen, + 'first_seen': None if not record else record.first_seen, + 'active_at': None if not record else record.active_at, + 'first_release_id': None if not record else record.first_release_id, }