diff --git a/CHANGELOG.md b/CHANGELOG.md index 37d25c6d..ece8bf65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,9 @@ List of the most important changes for each release. - Children of a dirty parent that is left out of the deserialization run are now deserialized against the parent's existing app row - For models whose `clean_fields` loads related objects, a dirty self-referential chain is now deserialized one level per run - Logs deserialization warnings under `morango.sync.stream.deserialize` instead of `morango.sync.operations` +- Removes `SyncableModel.deferred_clean_fields` and `morango.models.core.ForeignKeyReference` +- Removes `morango.sync.backends.utils.TemporaryTable` and `morango.sync.backends.utils.get_pk_field` +- Removes the unused bulk write helpers `_bulk_full_record_upsert`, `_bulk_update` and `_bulk_insert`, and the temporary table helper `_create_temporary_table`, from the sync backends ## 0.8.15 - Improves buffer serialization performance by performing bulk counter lookup diff --git a/morango/models/core.py b/morango/models/core.py index 319e242e..75fa500d 100644 --- a/morango/models/core.py +++ b/morango/models/core.py @@ -2,8 +2,6 @@ import json import logging import uuid -from collections import defaultdict -from collections import namedtuple from functools import reduce from django.core import exceptions @@ -745,9 +743,6 @@ class RecordMaxCounterBuffer(AbstractCounter): model_uuid = UUIDField(db_index=True) -ForeignKeyReference = namedtuple("ForeignKeyReference", ["from_field", "from_pk", "to_pk"]) - - class SyncableModel(UUIDModelMixin): """ ``SyncableModel`` is the base model class for syncing. Other models inherit from this class if they want to make @@ -873,39 +868,6 @@ def cached_clean_fields(self, fk_lookup_cache, exclude=None, sync_filter=None): ) fk_lookup_cache[self_key] = 1 - def deferred_clean_fields(self, exclude=None, sync_filter=None): - """ - Calls `.clean_fields()` but excludes all foreign key fields and instead returns them as a - dictionary for deferred batch processing - - :param exclude: A list of field names to exclude from validation - :type exclude: list[str] - :param sync_filter: The current sync's filter, if any - :type sync_filter: Filter|None - :return: A dictionary containing lists of `ForeignKeyReference`s keyed by the name of the - model being referenced by the FK - """ - excluded_fields = exclude or [] - deferred_fks = defaultdict(list) - for field in self._meta.fields: - if not isinstance(field, models.ForeignKey): - continue - # by not excluding the field if it's null, the default validation logic will apply - # and should raise a ValidationError if the FK field is not nullable - if getattr(self, field.attname) is None: - continue - excluded_fields.append(field.name) - deferred_fks[field.related_model._meta.verbose_name].append( - ForeignKeyReference( - from_field=field.attname, - from_pk=self.pk, - to_pk=getattr(self, field.attname), - ) - ) - - self.clean_fields(exclude=excluded_fields, sync_filter=sync_filter) - return deferred_fks - def serialize(self): """All concrete fields of the ``SyncableModel`` subclass, except for those specifically blacklisted, are returned in a dict.""" # NOTE: code adapted from https://github.com/django/django/blob/master/django/forms/models.py#L75 diff --git a/morango/sync/backends/base.py b/morango/sync/backends/base.py index c7ebda02..23ce93d2 100644 --- a/morango/sync/backends/base.py +++ b/morango/sync/backends/base.py @@ -7,8 +7,6 @@ class BaseSQLWrapper(object): - create_temporary_table_template = "CREATE TEMP TABLE {name} ({fields})" - def __init__(self, connection): self.connection = connection @@ -25,31 +23,6 @@ def _set_transaction_repeatable_read(self): """Set the current transaction isolation level""" yield - def _create_placeholder_list(self, fields, db_values): - # number of rows to update - num_of_rows = len(db_values) // len(fields) - # create '%s' placeholders for a single row - placeholder_tuple = tuple(["%s" for _ in range(len(fields))]) - # create list of the '%s' tuple placeholders based on number of rows to update - return [str(placeholder_tuple) for _ in range(num_of_rows)] - - def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values): - raise NotImplementedError("Subclass must implement this method.") - - def _bulk_insert(self, cursor, table_name, fields, db_values): - placeholder_str = ", ".join(self._create_placeholder_list(fields, db_values)).replace( - "'", "" - ) - fields_str = str(tuple(str(f.attname) for f in fields)).replace("'", "") - insert = """ - INSERT INTO {table_name} {fields} - VALUES {placeholder_str} - """.format(table_name=table_name, fields=fields_str, placeholder_str=placeholder_str) - cursor.execute(insert, db_values) - - def _bulk_update(self, cursor, table_name, fields, db_values): - raise NotImplementedError("Subclass must implement this method.") - def _dequeuing_delete_rmcb_records(self, cursor, transfersession_id): # delete all RMCBs which are a reverse FF (store version newer than buffer version) delete_rmcb_records = """DELETE FROM {rmcb} @@ -176,16 +149,6 @@ def _dequeuing_delete_remaining_buffer(self, cursor, transfersession_id): ) cursor.execute(delete_remaining_buffer) - def _create_temporary_table(self, cursor, name, field_sqls, fields_params): - """ - :param cursor: The database connection cursor - :param name: The str name of the temp table - :param field_sqls: A list of SQL strings representing the fields - :param fields_params: A list of SQL parameters if necessary for the fields SQL - """ - sql = self.create_temporary_table_template.format(name=name, fields=", ".join(field_sqls)) - cursor.execute(sql, fields_params) - def _lock_all_partitions(self, shared=False): """ Execute a lock within the database for all partitions, if the backend supports it. The lock diff --git a/morango/sync/backends/postgres.py b/morango/sync/backends/postgres.py index 04afc809..29dfffe4 100644 --- a/morango/sync/backends/postgres.py +++ b/morango/sync/backends/postgres.py @@ -10,7 +10,6 @@ from morango.utils import SETTINGS from .base import BaseSQLWrapper -from .utils import get_pk_field # advisory lock integers for locking partitions LOCK_ALL_PARTITIONS = 1 @@ -23,7 +22,6 @@ class SQLWrapper(BaseSQLWrapper): backend = "postgresql" - create_temporary_table_template = "CREATE TEMP TABLE {name} ({fields}) ON COMMIT DROP" def _is_transaction_isolation_error(self, error): """ @@ -85,96 +83,6 @@ def _set_transaction_repeatable_read(self): self.connection.connection.set_session(isolation_level=existing_isolation_level) self.connection.connection.set_session(autocommit=existing_autocommit) - def _prepare_with_values(self, name, fields, db_values): - placeholder_list = self._create_placeholder_list(fields, db_values) - # convert this list to a string to be passed into raw sql query - placeholder_str = ", ".join(placeholder_list).replace("'", "") - return """ - WITH {name} {fields} as - ( - VALUES {placeholder_str} - ) - """.format( - name=name, - fields=str(tuple(str(f.column) for f in fields)).replace("'", ""), - placeholder_str=placeholder_str, - ) - - def _prepare_casted_fields(self, fields): - return ", ".join( - map( - lambda f: "{f}::{type}".format(f=f.column, type=f.rel_db_type(self.connection)), - fields, - ) - ) - - def _prepare_set_casted_values(self, fields, source_table): - return ", ".join( - map( - lambda f: "{f} = {src}.{f}::{type}".format( - f=f.attname, - type=f.rel_db_type(self.connection), - src=source_table, - ), - fields, - ) - ) - - def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values): - pk = get_pk_field(fields) - - cte_name = "new_values" - upsert = """ - {cte}, - updated as - ( - UPDATE {table_name} model - SET {set_values} - FROM {cte_name} cte - WHERE model.id = cte.{pk_field}::{pk_type} - RETURNING model.{pk_field} - ) - INSERT INTO {table_name} {fields} - SELECT {select_fields} - FROM {cte_name} cte - WHERE cte.{pk_field}::{pk_type} NOT IN (SELECT {pk_field} FROM updated) - """.format( - cte=self._prepare_with_values(cte_name, fields, db_values), - cte_name=cte_name, - table_name=table_name, - fields=str(tuple(str(f.column) for f in fields)).replace("'", ""), - set_values=self._prepare_set_casted_values(fields, "cte"), - select_fields=self._prepare_casted_fields(fields), - pk_field=pk.column, - pk_type=pk.rel_db_type(self.connection), - ) - # use DB-APIs parameter substitution (2nd parameter expects a sequence) - cursor.execute(upsert, db_values) - - def _bulk_update(self, cursor, table_name, fields, db_values): - pk = get_pk_field(fields) - - insert = """ - {cte} - UPDATE {table_name} model - SET {set_values} - FROM {cte_name} cte - WHERE model.{pk_field} = cte.{pk_field}::{pk_type} - """ - - cte_name = "new_values" - insert = insert.format( - cte=self._prepare_with_values(cte_name, fields, db_values), - cte_name=cte_name, - table_name=table_name, - fields=str(tuple(str(f.column) for f in fields)).replace("'", ""), - set_values=self._prepare_set_casted_values(fields, "cte"), - pk_field=pk.column, - pk_type=pk.rel_db_type(self.connection), - ) - # use DB-APIs parameter substitution (2nd parameter expects a sequence) - cursor.execute(insert, db_values) - def _dequeuing_merge_conflict_rmcb(self, cursor, transfersession_id): # transfer record max counters for records with merge conflicts + perform max merge_conflict_rmc = """UPDATE {rmc} as rmc SET counter diff --git a/morango/sync/backends/sqlite.py b/morango/sync/backends/sqlite.py index e6cfd307..8b98cf4e 100644 --- a/morango/sync/backends/sqlite.py +++ b/morango/sync/backends/sqlite.py @@ -4,104 +4,11 @@ from morango.models.core import Store from .base import BaseSQLWrapper -from .utils import calculate_max_sqlite_variables -from .utils import get_pk_field class SQLWrapper(BaseSQLWrapper): backend = "sqlite" - def _bulk_full_record_upsert(self, cursor, table_name, fields, db_values): - """ - Example query: - `REPLACE INTO model (F1,F2,F3) VALUES (%s, %s, %s), (%s, %s, %s), (%s, %s, %s)` - where values=[1,2,3,4,5,6,7,8,9] - """ - placeholder_list = self._create_placeholder_list(fields, db_values) - # calculate and create equal sized chunks of data to insert incrementally - num_of_rows_able_to_insert = calculate_max_sqlite_variables() // len(fields) - num_of_values_able_to_insert = num_of_rows_able_to_insert * len(fields) - value_chunks = [ - db_values[x : x + num_of_values_able_to_insert] - for x in range(0, len(db_values), num_of_values_able_to_insert) - ] - placeholder_chunks = [ - placeholder_list[x : x + num_of_rows_able_to_insert] - for x in range(0, len(placeholder_list), num_of_rows_able_to_insert) - ] - # insert data chunks - fields_str = str(tuple(str(f.attname) for f in fields)).replace("'", "") - for values, params in zip(value_chunks, placeholder_chunks): - placeholder_str = ", ".join(params).replace("'", "") - insert = """ - REPLACE INTO {table_name} {fields} - VALUES {placeholder_str} - """.format( - table_name=table_name, - fields=fields_str, - placeholder_str=placeholder_str, - ) - # use DB-APIs parameter substitution (2nd parameter expects a sequence) - cursor.execute(insert, values) - - def _bulk_insert(self, cursor, table_name, fields, db_values): - num_of_rows_able_to_insert = calculate_max_sqlite_variables() // len(fields) - num_of_values_able_to_insert = num_of_rows_able_to_insert * len(fields) - value_chunks = [ - db_values[x : x + num_of_values_able_to_insert] - for x in range(0, len(db_values), num_of_values_able_to_insert) - ] - for value_chunk in value_chunks: - super(SQLWrapper, self)._bulk_insert(cursor, table_name, fields, value_chunk) - - def _bulk_update(self, cursor, table_name, fields, db_values): - """ - Example query: - `UPDATE model SET F1=(CASE id WHEN %s THEN %s ... END) ...` - WHERE id IN [...] - """ - # calculate and create equal sized chunks of data to update incrementally - # for every field we're updating, we'll require 3 parameters - num_update_fields = len(fields) - 1 - num_of_rows_able_to_update = calculate_max_sqlite_variables() // num_update_fields // 3 - num_of_values_able_to_update = num_of_rows_able_to_update * len(fields) - value_chunks = [ - db_values[x : x + num_of_values_able_to_update] - for x in range(0, len(db_values), num_of_values_able_to_update) - ] - pk = get_pk_field(fields) - - # insert data chunks - for values in value_chunks: - set_sql = "" - params = [] - pk_params = [] - for field in fields: - if field == pk: - continue - set_field_sql = " {field} = (CASE {pk_field}".format( - field=field.column, pk_field=pk.column - ) - for y in range(0, len(values), len(fields)): - value_set = values[y : y + len(fields)] - set_field_sql += " WHEN %s THEN %s" - pk_params.append(value_set[fields.index(pk)]) - params.append(value_set[fields.index(pk)]) - params.append(value_set[fields.index(field)]) - set_field_sql += " END)," - set_sql += set_field_sql - params.extend(pk_params) - update = """ - UPDATE {table_name} SET {set_sql} WHERE {pk_field} IN {placeholder_str} - """.format( - table_name=table_name, - set_sql=set_sql[:-1], - pk_field=pk.column, - placeholder_str="({})".format(",".join("%s" for _ in range(len(pk_params)))), - ) - # use DB-APIs parameter substitution (2nd parameter expects a sequence) - cursor.execute(update, params) - def _dequeuing_merge_conflict_rmcb(self, cursor, transfersession_id): # transfer record max counters for records with merge conflicts + perform max merge_conflict_rmc = """REPLACE INTO {rmc} (instance_id, counter, store_model_id) diff --git a/morango/sync/backends/utils.py b/morango/sync/backends/utils.py index 527dc542..d3aa769b 100644 --- a/morango/sync/backends/utils.py +++ b/morango/sync/backends/utils.py @@ -23,7 +23,7 @@ def calculate_max_sqlite_variables(): """ SQLite has a limit on the max number of variables allowed for parameter substitution. This limit used to be 999, but can be compiled to a different number, and is now often much larger. This function reads the value from the compile options. - We use this value to chunk our SQL bulk insert statements when deserializing from the store to the app layer. + Callers use this value to chunk SQL statements so they stay within the limit. """ conn = sqlite3.connect(":memory:") @@ -47,100 +47,3 @@ def calculate_max_sqlite_variables(): break return MAX_VARIABLE_NUMBER - - -def get_pk_field(fields): - try: - return next(f for f in fields if f.primary_key) - except StopIteration: - raise ValueError("No primary key found in fields") - - -class TemporaryTable(object): - """ - Utility class for managing a temporary table within the database - """ - - __slots__ = ("connection", "name", "fields", "backend", "_meta") - - def __init__(self, connection, name, **fields): - """ - :param connection: A database connection object - :param name: A str name for the table in the database - :param fields: Keyword arguments are assumed to be fields for defining the schema of the - temporary table - """ - self.connection = connection - self.name = name - self.fields = [] - self.backend = load_backend(connection) - self._meta = self.Meta() - - for name, field in fields.items(): - field.set_attributes_from_name(name) - self.fields.append(field) - - def __enter__(self): - self.create() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self.drop() - - @property - def sql_name(self): - """ - :return: The name actual name of the table used in the DB, prefixed to avoid collisions - """ - return self.connection.ops.quote_name("t_{}".format(self.name)) - - def get_field(self, name): - """ - :param name: A str of the name of which field to find - :return: The field object - """ - return next(f for f in self.fields if f.name == name) - - def create(self): - """ - Creates the temporary table within the database - """ - fields = [] - params = [] - schema_editor = self.connection.schema_editor() - for field in self.fields: - # generates the SQL expression for the table column - field_sql, field_params = schema_editor.column_sql(self, field, include_default=True) - field_sql_name = self.connection.ops.quote_name(field.column) - fields.append("{name} {sql}".format(name=field_sql_name, sql=field_sql)) - params.extend(field_params) - with self.connection.cursor() as c: - self.backend._create_temporary_table(c, self.sql_name, fields, params) - - def drop(self): - """ - Drops the temporary table within the database - """ - with self.connection.cursor() as c: - c.execute("DROP TABLE IF EXISTS {name}".format(name=self.sql_name)) - - def bulk_insert(self, values): - """ - Bulk inserts a list of records into the temporary table - - :param values: A list of dictionaries containing data to insert, keyed by field name - """ - params = [] - for value_dict in values: - for field in self.fields: - params.append(value_dict.get(field.attname)) - with self.connection.cursor() as c: - self.backend._bulk_insert(c, self.sql_name, self.fields, params) - - class Meta: - """ - HACK: some Django code bits require a model, only to access meta information, so we use - this to mimic a model class - """ - - db_tablespace = None diff --git a/tests/testapp/tests/models/test_core.py b/tests/testapp/tests/models/test_core.py index cdc68d2f..74713782 100644 --- a/tests/testapp/tests/models/test_core.py +++ b/tests/testapp/tests/models/test_core.py @@ -373,13 +373,6 @@ def test_cached_clean_fields(self, mock_clean_fields): exclude=["test1", "parent"], sync_filter=sync_filter ) - @mock.patch("morango.models.core.SyncableModel.clean_fields") - def test_deferred_clean_fields(self, mock_clean_fields): - f = Facility(name="test") - sync_filter = Filter("test") - f.deferred_clean_fields(exclude=["test1"], sync_filter=sync_filter) - mock_clean_fields.assert_called_once_with(exclude=["test1"], sync_filter=sync_filter) - class StoreDeserializationErrorTestCase(TestCase): """Tests for Store.set_deserialization_error() and Store.unset_deserialization_error()"""