diff --git a/core/importers/models.py b/core/importers/models.py index ce99e80e4..e7673be9a 100644 --- a/core/importers/models.py +++ b/core/importers/models.py @@ -109,12 +109,13 @@ class BaseResourceImporter: mandatory_fields = set() allowed_fields = [] - def __init__(self, data, user, update_if_exists=False): + def __init__(self, data, user, update_if_exists=False, cache=None): self.user = user self.data = data self.update_if_exists = update_if_exists self.queryset = None self.index_resources = False + self.cache = cache if cache is not None else {} @classmethod def can_handle(cls, obj): @@ -127,6 +128,15 @@ def get_resource_type(): def get(self, attr, default_value=None): return self.data.get(attr, default_value) + def get_cached_parent_source(self): + cache = self.cache.setdefault('source_by_owner', {}) + key = (self.get_owner_type_filter(), self.get('owner'), self.get('source')) + if key not in cache: + cache[key] = Source.objects.filter( + **{self.get_owner_type_filter(): self.get('owner')}, mnemonic=self.get('source'), version=HEAD + ).first() + return cache[key] + def parse(self): self.data = self.get_filter_allowed_fields() self.data['created_by'] = self.data['updated_by'] = self.user @@ -416,8 +426,8 @@ class ConceptImporter(BaseResourceImporter): def get_resource_type(): return 'Concept' - def __init__(self, data, user, update_if_exists, skip_hierarchy_tasks=False): - super().__init__(data, user, update_if_exists) + def __init__(self, data, user, update_if_exists, skip_hierarchy_tasks=False, cache=None): # pylint: disable=too-many-arguments + super().__init__(data, user, update_if_exists, cache=cache) self.skip_hierarchy_tasks = skip_hierarchy_tasks self.version = False self.instance = None @@ -439,9 +449,7 @@ def get_queryset(self): return self.queryset def parse(self): - source = Source.objects.filter( - **{self.get_owner_type_filter(): self.get('owner')}, mnemonic=self.get('source'), version=HEAD - ).first() + source = self.get_cached_parent_source() super().parse() self.data['parent'] = source self.data['mnemonic'] = str(self.data.pop('id', '')) @@ -527,14 +535,29 @@ class MappingImporter(BaseResourceImporter): def get_resource_type(): return 'Mapping' - def __init__(self, data, user, update_if_exists): - super().__init__(data, user, update_if_exists) + def __init__(self, data, user, update_if_exists, cache=None): + super().__init__(data, user, update_if_exists, cache=cache) self.version = False self.instance = None def exists(self): return self.get_queryset().exists() + def get_cached_versioned_concept_id_by_uri(self, uri): + # Only versioned_object_id is ever read from the resolved concept here, so cache just that + # (and fetch just that) instead of retaining full Concept instances for the chunk's lifetime. + cache = self.cache.setdefault('concept_versioned_id_by_uri', {}) + if uri not in cache: + cache[uri] = Concept.objects.filter(id=F('versioned_object_id'), uri=uri).values_list( + 'versioned_object_id', flat=True).first() + return cache[uri] + + def get_cached_source_exists_by_uri(self, uri): + cache = self.cache.setdefault('source_exists_by_uri', {}) + if uri not in cache: + cache[uri] = Source.objects.filter(uri=uri).exists() + return cache[uri] + def get_queryset(self): # pylint: disable=too-many-branches if self.queryset: return self.queryset @@ -556,21 +579,21 @@ def get_queryset(self): # pylint: disable=too-many-branches ] versionless_from_concept_url = drop_version(from_concept_url) - from_concept = Concept.objects.filter(id=F('versioned_object_id'), uri=versionless_from_concept_url).first() - if from_concept: - filters['from_concept__versioned_object_id'] = from_concept.versioned_object_id + from_concept_id = self.get_cached_versioned_concept_id_by_uri(versionless_from_concept_url) + if from_concept_id is not None: + filters['from_concept__versioned_object_id'] = from_concept_id elif not from_concept_code: filters['from_concept_code'] = compact(versionless_from_concept_url.split('/'))[-1] if to_concept_url: versionless_to_concept_url = drop_version(to_concept_url) - to_concept = Concept.objects.filter(id=F('versioned_object_id'), uri=versionless_to_concept_url).first() - if to_concept: - filters['to_concept__versioned_object_id'] = to_concept.versioned_object_id + to_concept_id = self.get_cached_versioned_concept_id_by_uri(versionless_to_concept_url) + if to_concept_id is not None: + filters['to_concept__versioned_object_id'] = to_concept_id else: filters['to_concept_code'] = compact(versionless_to_concept_url.split('/'))[-1] if not to_source_url: to_source_uri = to_parent_uri(versionless_to_concept_url) - if Source.objects.filter(uri=drop_version(to_source_uri)).exists(): + if self.get_cached_source_exists_by_uri(drop_version(to_source_uri)): filters['to_source__uri'] = to_source_uri if self.get('id'): @@ -589,9 +612,7 @@ def get_queryset(self): # pylint: disable=too-many-branches return self.queryset def parse(self): - source = Source.objects.filter( - **{self.get_owner_type_filter(): self.get('owner')}, mnemonic=self.get('source'), version=HEAD - ).first() + source = self.get_cached_parent_source() self.data = self.get_filter_allowed_fields() self.data['parent'] = source @@ -631,14 +652,15 @@ def process(self): self.instance = queryset.first().clone() self.instance._counted = None # pylint: disable=protected-access self.instance._index = False # pylint: disable=protected-access - errors = Mapping.create_new_version_for(self.instance, self.data, self.user) + errors = Mapping.create_new_version_for(self.instance, self.data, self.user, cache=self.cache) if errors and Mapping.is_standard_checksum_error(errors): return UNCHANGED return errors or UPDATED if 'update_comment' in self.data: self.data['comment'] = self.data['update_comment'] self.data.pop('update_comment') - self.instance = Mapping.persist_new({**self.data, '_counted': None, '_index': False}, self.user) + self.instance = Mapping.persist_new( + {**self.data, '_counted': None, '_index': False}, self.user, cache=self.cache) if self.instance.id: return CREATED return self.instance.errors or errors or FAILED @@ -752,6 +774,8 @@ def delete(self): # pylint: disable=too-many-locals,too-many-branches class BulkImportInline(BaseImporter): + PROGRESS_NOTIFY_INTERVAL_SECONDS = 2 + def __init__( # pylint: disable=too-many-arguments self, content, username, update_if_exists=False, input_list=None, user=None, set_user=True, self_task_id=None, skip_hierarchy_tasks=False @@ -759,6 +783,15 @@ def __init__( # pylint: disable=too-many-arguments super().__init__(content, username, update_if_exists, user, not bool(input_list), set_user) self.self_task_id = self_task_id self.skip_hierarchy_tasks = skip_hierarchy_tasks + # Lookup cache shared across this run's items (see ConceptImporter/MappingImporter). + # It caches misses too (None/False), so it's only safe because a chunk is single-resource-type + # in the production parallel path (BulkImportParallelRunner.make_parts splits concepts and + # mappings into separate chunks) -- a mapping chunk never creates the concepts it resolves, and + # a concept chunk's parent source already exists. If a future change interleaves resource types + # within one BulkImportInline run (e.g. the deprecated BulkImportInlineView), a mapping could + # cache a "not found" for a concept created earlier in the same run. Don't share this cache + # across resource types unless that invariant is re-verified. + self.cache = {} self.set_task() if input_list: self.input_list = input_list @@ -777,6 +810,7 @@ def __init__( # pylint: disable=too-many-arguments self.processed = 0 self.total = len(self.input_list) self.start_time = time.time() + self.last_progress_notified_at = 0 self.elapsed_seconds = 0 self.index_resources = False @@ -819,21 +853,26 @@ def handle_item_import_result(self, result, item): # pylint: disable=too-many-r print("****Unexpected Result****", result) self.others.append(item) - def notify_progress(self): - if self.task: # pragma: no cover - self.task.summary = { - 'total': self.total, - 'processed': self.processed, - 'created': len(self.created), - 'updated': len(self.updated), - 'invalid': len(self.invalid), - 'failed': len(self.failed), - 'deleted': len(self.deleted), - 'not_found': len(self.not_found), - 'permission_denied': len(self.permission_denied), - 'unchanged': len(self.unchanged), - } - self.task.save() + def notify_progress(self, force=False): + if not self.task: + return + now = time.time() + if not force and (now - self.last_progress_notified_at) < self.PROGRESS_NOTIFY_INTERVAL_SECONDS: + return + self.last_progress_notified_at = now + self.task.summary = { # pragma: no cover + 'total': self.total, + 'processed': self.processed, + 'created': len(self.created), + 'updated': len(self.updated), + 'invalid': len(self.invalid), + 'failed': len(self.failed), + 'deleted': len(self.deleted), + 'not_found': len(self.not_found), + 'permission_denied': len(self.permission_denied), + 'unchanged': len(self.unchanged), + } + self.task.save() def run(self): # pylint: disable=too-many-branches,too-many-statements,too-many-locals if self.self_task_id: # pragma: no cover @@ -883,7 +922,8 @@ def run(self): # pylint: disable=too-many-branches,too-many-statements,too-many try: concept_importer = ConceptImporter( item, self.user, self.update_if_exists, - skip_hierarchy_tasks=self.skip_hierarchy_tasks and bool(item.get('id')) + skip_hierarchy_tasks=self.skip_hierarchy_tasks and bool(item.get('id')), + cache=self.cache ) _result = concept_importer.delete() if action == 'delete' else concept_importer.run() if self.index_resources and get(concept_importer.instance, 'id'): @@ -902,7 +942,7 @@ def run(self): # pylint: disable=too-many-branches,too-many-statements,too-many continue if item_type == 'mapping': try: - mapping_importer = MappingImporter(item, self.user, self.update_if_exists) + mapping_importer = MappingImporter(item, self.user, self.update_if_exists, cache=self.cache) _result = mapping_importer.delete() if action == 'delete' else mapping_importer.run() if self.index_resources and get(mapping_importer.instance, 'id'): new_mapping_ids.update(set(compact( @@ -925,6 +965,7 @@ def run(self): # pylint: disable=too-many-branches,too-many-statements,too-many ) continue + self.notify_progress(force=True) if new_concept_ids: for chunk in chunks(list(set(new_concept_ids)), 5000): batch_index_resources.apply_async( diff --git a/core/importers/tests.py b/core/importers/tests.py index a1de15503..bbb8ef1f4 100644 --- a/core/importers/tests.py +++ b/core/importers/tests.py @@ -712,6 +712,123 @@ def test_mapping_import(self, batch_index_resources_mock): # pylint: disable=to self.assertEqual(importer.failed[0]['errors'], {'source': 'Not Found'}) self.assertTrue(importer.elapsed_seconds > 0) + @patch('core.importers.models.batch_index_resources') + def test_mapping_import_cache_reuse_correctness(self, batch_index_resources_mock): + # Several mappings in the same BulkImportInline run sharing the same from/to concepts -- + # mirrors the production case (a mapping chunk resolving concepts that already exist) where + # the lookup cache introduced for performance should be transparent to correctness: every + # mapping must resolve to the same from_concept/to_concept regardless of cache hits. + batch_index_resources_mock.__name__ = 'batch_index_resources' + source = OrganizationSourceFactory( + organization=(OrganizationFactory(mnemonic='DemoOrg')), mnemonic='DemoSource', version='HEAD' + ) + vegetable = ConceptFactory(parent=source, mnemonic='Vegetable') + corn = ConceptFactory(parent=source, mnemonic='Corn') + carrot = ConceptFactory(parent=source, mnemonic='Carrot') + + input_list = [ + { + "to_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Corn/", + "from_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Vegetable/", + "type": "Mapping", "source": "DemoSource", + "owner": "DemoOrg", "map_type": "Has Child", "owner_type": "Organization", + }, + { + "to_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Carrot/", + "from_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Vegetable/", + "type": "Mapping", "source": "DemoSource", + "owner": "DemoOrg", "map_type": "Has Child", "owner_type": "Organization", + }, + { + "to_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Corn/", + "from_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Vegetable/", + "type": "Mapping", "source": "DemoSource", + "owner": "DemoOrg", "map_type": "Same As", "owner_type": "Organization", + }, + ] + + importer = BulkImportInline(content=None, username='ocladmin', update_if_exists=True, input_list=input_list) + importer.run() + + self.assertEqual(importer.processed, 3) + self.assertEqual(len(importer.created), 3) + self.assertEqual(importer.failed, []) + + has_child_corn = Mapping.objects.filter( + map_type='Has Child', id=F('versioned_object_id'), to_concept_code='Corn' + ).first() + has_child_carrot = Mapping.objects.filter( + map_type='Has Child', id=F('versioned_object_id'), to_concept_code='Carrot' + ).first() + same_as_corn = Mapping.objects.filter( + map_type='Same As', id=F('versioned_object_id'), to_concept_code='Corn' + ).first() + + self.assertEqual(has_child_corn.from_concept_id, vegetable.id) + self.assertEqual(has_child_corn.to_concept_id, corn.id) + self.assertEqual(has_child_carrot.from_concept_id, vegetable.id) + self.assertEqual(has_child_carrot.to_concept_id, carrot.id) + self.assertEqual(same_as_corn.from_concept_id, vegetable.id) + self.assertEqual(same_as_corn.to_concept_id, corn.id) + + # cache was actually exercised and shared across items, not just present and unused + self.assertEqual(len(importer.cache['source_by_owner']), 1) + self.assertEqual(len(importer.cache['concept_versioned_id_by_uri']), 3) # Vegetable, Corn, Carrot + self.assertEqual(len(importer.cache['concept_by_expr']), 3) + + @patch('core.importers.models.batch_index_resources') + def test_mapping_import_cache_stale_after_concept_created_in_same_run(self, batch_index_resources_mock): + # Documents the invariant called out on BulkImportInline.cache: the lookup cache stores + # misses too, so if a resource type is created mid-run and an earlier item in the *same* run + # already cached its absence, a later item referencing it can get a stale miss. This is safe + # in production because BulkImportParallelRunner only ever puts one resource type per chunk + # (a mapping chunk never creates the concepts it resolves) -- this test exists so that + # invariant has an executable tripwire if chunking ever changes to interleave resource types. + batch_index_resources_mock.__name__ = 'batch_index_resources' + source = OrganizationSourceFactory( + organization=(OrganizationFactory(mnemonic='DemoOrg')), mnemonic='DemoSource', version='HEAD' + ) + ConceptFactory(parent=source, mnemonic='Vegetable') + + input_list = [ + { + "to_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Sugar/", + "from_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Vegetable/", + "type": "Mapping", "source": "DemoSource", + "owner": "DemoOrg", "map_type": "Has Child", "owner_type": "Organization", + }, + { + "id": "Sugar", "type": "Concept", "concept_class": "Misc", "datatype": "None", + "source": "DemoSource", "owner": "DemoOrg", "owner_type": "Organization", + "names": [ + {"name": "Sugar", "locale": "en", "locale_preferred": "True", "name_type": "Fully Specified"}], + }, + { + "to_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Sugar/", + "from_concept_url": "/orgs/DemoOrg/sources/DemoSource/concepts/Vegetable/", + "type": "Mapping", "source": "DemoSource", + "owner": "DemoOrg", "map_type": "Same As", "owner_type": "Organization", + }, + ] + + importer = BulkImportInline(content=None, username='ocladmin', update_if_exists=True, input_list=input_list) + importer.run() + + sugar = Concept.objects.filter(mnemonic='Sugar', id=F('versioned_object_id')).first() + self.assertIsNotNone(sugar) + + same_as_sugar = Mapping.objects.filter( + map_type='Same As', id=F('versioned_object_id') + ).first() + self.assertIsNotNone(same_as_sugar) + # Known limitation of the invariant above: the third item's lookup hits the stale cached + # miss from the first item (Sugar didn't exist yet when that mapping resolved it), so the + # mapping is created without a linked to_concept even though Sugar exists by this point. If + # this assertion ever starts failing because to_concept_id became non-null, the cache started + # being invalidated correctly (or chunking changed) -- update this test and the comment on + # BulkImportInline.cache rather than treating the new behavior as a regression. + self.assertIsNone(same_as_sugar.to_concept_id) + @patch('core.importers.models.batch_index_resources') def test_mapping_import_with_autoid_assignment(self, batch_index_resources_mock): self.assertEqual(Mapping.objects.count(), 0) diff --git a/core/mappings/models.py b/core/mappings/models.py index ea9852346..3b994b449 100644 --- a/core/mappings/models.py +++ b/core/mappings/models.py @@ -317,7 +317,7 @@ def create_initial_version(cls, mapping, **kwargs): initial_version.save() return initial_version - def populate_fields_from_relations(self, data): # pylint: disable=too-many-locals + def populate_fields_from_relations(self, data, cache=None): # pylint: disable=too-many-locals from core.concepts.models import Concept from core.sources.models import Source @@ -328,19 +328,33 @@ def populate_fields_from_relations(self, data): # pylint: disable=too-many-loca from_source_url = data.get('from_source_url', None) or to_parent_uri(from_concept_url) to_source_url = data.get('to_source_url', None) or to_parent_uri(to_concept_url) + concept_cache = None if cache is None else cache.setdefault('concept_by_expr', {}) + source_cache = None if cache is None else cache.setdefault('source_resolve_ref', {}) + def get_concept(expr): if expr and not expr.endswith('/'): expr = expr + '/' + if concept_cache is not None and expr in concept_cache: + return concept_cache[expr] concept = Concept.objects.filter( uri=expr).first() or Concept.objects.filter(uri=encode_string(expr, safe='/')).first() - return concept or {'mnemonic': expr.replace(to_parent_uri(expr), '').replace('concepts/', '').split('/')[0]} + result = concept or {'mnemonic': expr.replace(to_parent_uri(expr), '').replace('concepts/', '').split('/')[0]} + if concept_cache is not None: + concept_cache[expr] = result + return result def get_source(url): + if source_cache is not None and url in source_cache: + return source_cache[url] source, _ = Source.resolve_reference_expression(url, None, HEAD) if source.id: - return source, source.versioned_object_url or source.resolution_url or url - return None, source.resolution_url or url + result = (source, source.versioned_object_url or source.resolution_url or url) + else: + result = (None, source.resolution_url or url) + if source_cache is not None: + source_cache[url] = result + return result self.from_source, self.from_source_url = get_source(from_source_url) self.to_source, self.to_source_url = get_source(to_source_url) @@ -392,8 +406,8 @@ def latest_source_version(self): return self.sources.exclude(version=HEAD).order_by('-created_at').first() @classmethod - def create_new_version_for(cls, instance, data, user): - instance.populate_fields_from_relations(data) + def create_new_version_for(cls, instance, data, user, cache=None): + instance.populate_fields_from_relations(data, cache=cache) instance.extras = data.get('extras', instance.extras) instance.external_id = data.get('external_id', instance.external_id) instance.mnemonic = data.get('mnemonic', instance.mnemonic) @@ -464,7 +478,7 @@ def get_next_sort_weight(self): return None @classmethod - def persist_new(cls, data, user): # pylint: disable=too-many-statements + def persist_new(cls, data, user, cache=None): # pylint: disable=too-many-statements related_fields = ['from_concept_url', 'to_concept_url', 'to_source_url', 'from_source_url'] field_data = {k: v for k, v in data.items() if k not in related_fields} url_params = {k: v for k, v in data.items() if k in related_fields} @@ -478,7 +492,7 @@ def persist_new(cls, data, user): # pylint: disable=too-many-statements if mapping.is_existing_in_parent(): mapping.errors = {'__all__': [ALREADY_EXISTS]} return mapping - mapping.populate_fields_from_relations(url_params) + mapping.populate_fields_from_relations(url_params, cache=cache) try: mapping.full_clean() diff --git a/tools/benchmark_bulk_import.py b/tools/benchmark_bulk_import.py new file mode 100644 index 000000000..86f595559 --- /dev/null +++ b/tools/benchmark_bulk_import.py @@ -0,0 +1,102 @@ +""" +Benchmarks the default (non-S3) bulk import path against a sample file, +reporting wall-clock time and total DB query count. + +Runs inside a transaction that is always rolled back at the end, so it is +safe to run against a real dev DB -- nothing it creates is kept. + +Usage (inside the api container, or any env with DJANGO_SETTINGS_MODULE set): + python tools/benchmark_bulk_import.py [path/to/sample.json] [--username ocladmin] + +Defaults to core/samples/pepfar_datim_moh_fy19.json (413 mixed concept/mapping/ +reference rows) since it's already exercised by core/importers/tests.py and +needs no extra fixtures beyond the 'ocladmin' user that ships with every env. + +To compare old vs. new behavior, run this once on the current code, then +`git stash` the importer changes and run it again, e.g.: + python tools/benchmark_bulk_import.py # with changes + git stash && python tools/benchmark_bulk_import.py && git stash pop # baseline +""" +import argparse +import os +import sys +import time +from collections import deque + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings') + +import django # noqa: E402 pylint: disable=wrong-import-position + +django.setup() + +from django.db import transaction # noqa: E402 pylint: disable=wrong-import-position +from django.test.utils import CaptureQueriesContext # noqa: E402 pylint: disable=wrong-import-position +from django.db import connection # noqa: E402 pylint: disable=wrong-import-position + +from core.importers.models import BulkImportInline # noqa: E402 pylint: disable=wrong-import-position + +DEFAULT_SAMPLE = os.path.join( + os.path.dirname(__file__), '..', 'core', 'samples', 'pepfar_datim_moh_fy19.json' +) + + +class RollbackTransaction(Exception): + pass + + +def run_import(content, username): + # Django's per-connection query log is capped at 9000 entries (deque maxlen); + # bulk imports blow past that easily, which would silently truncate the count. + connection.queries_log = deque(maxlen=10_000_000) + importer = BulkImportInline(content, username, True) + with CaptureQueriesContext(connection) as queries_ctx: + start = time.perf_counter() + importer.run() + elapsed = time.perf_counter() - start + return importer, elapsed, len(queries_ctx.captured_queries) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('sample', nargs='?', default=DEFAULT_SAMPLE) + parser.add_argument('--username', default='ocladmin') + args = parser.parse_args() + + with open(args.sample, 'r') as file: + content = file.read() + + print(f"Sample: {args.sample}") + print(f"Username: {args.username}") + print("Running import inside a transaction that will be rolled back...") + + importer = None + elapsed = None + query_count = None + try: + with transaction.atomic(): + importer, elapsed, query_count = run_import(content, args.username) + raise RollbackTransaction() + except RollbackTransaction: + pass + + print() + print("==== Result ====") + print(f"processed: {importer.processed}") + print(f"created: {len(importer.created)}") + print(f"updated: {len(importer.updated)}") + print(f"unchanged: {len(importer.unchanged)}") + print(f"invalid: {len(importer.invalid)}") + print(f"failed: {len(importer.failed)}") + print(f"permission_denied: {len(importer.permission_denied)}") + print() + print("==== Performance ====") + print(f"wall_time_seconds: {elapsed:.4f}") + print(f"db_query_count: {query_count}") + if importer.processed: + print(f"queries_per_item: {query_count / importer.processed:.2f}") + print(f"ms_per_item: {(elapsed * 1000) / importer.processed:.3f}") + + +if __name__ == '__main__': + sys.exit(main())