diff --git a/montage/rdb.py b/montage/rdb.py index db45356f..f0894377 100644 --- a/montage/rdb.py +++ b/montage/rdb.py @@ -654,6 +654,69 @@ def _entry_insert_row(entry): return {key: getattr(entry, key) for key in _ENTRY_INSERT_COLS} +# Imports have stored Commons' file_id since 2026-05-31: on production the +# last category-imported entry without one dates from 2026-05-30 13:21, the +# first with one from 2026-05-31 20:52. A file new to Montage after this date +# therefore has its file_id stored (hatnote/montage#513). +FILE_ID_LOOKUP_CUTOFF = datetime.datetime(2026, 6, 1) + +# Batch size for lookups by file_id: below 200 on purpose. MariaDB's +# eq_range_index_dive_limit is 200 by default (ToolsDB too): with fewer +# values in an IN (...) list the optimizer counts the matching rows in the +# index and uses ix_entries_file_id; from 200 values on it estimates from +# index statistics, misjudges this column and reads the whole table. +# Measured on production on 2026-10-05 (#513): 199 ids took 2-7 ms with the +# index, for existing and for new file_ids alike; 200 ids read the whole +# table (0.56 s warm, 9.4 s cold). If a future server or setting changes +# this (check with EXPLAIN), a fallback is +# query.with_hint(Entry, 'FORCE INDEX (ix_entries_file_id)', 'mysql'): +# 2 ms for 200 ids there, but only where that index exists, as the model +# does not declare it (tools/migrate_prod_db.sql creates it). +FILE_ID_LOOKUP_CHUNK_SIZE = 199 + + +def _use_file_id_lookup(rnd, entries): + """Whether add_entries looks entries up by file_id instead of by name + (#513): when the campaign opened after FILE_ID_LOOKUP_CUTOFF and every + entry has a file_id (wikireplica imports; CSV imports have none). + Production has no index on entries.name, so a lookup by name reads the + whole table; by file_id it uses ix_entries_file_id, in batches of + FILE_ID_LOOKUP_CHUNK_SIZE. + + Files not found by file_id are looked up by name only where entries.name + is indexed (see _add_entries_by_file_id); elsewhere a file imported + before the cutoff, without a file_id, gets a second entries row. + Campaigns opening after the cutoff rarely contain such files.""" + open_date = rnd.campaign.open_date if rnd.campaign else None + if open_date is None or open_date < FILE_ID_LOOKUP_CUTOFF: + return False + return bool(entries) and all(e.file_id is not None for e in entries) + + +_ENTRIES_INDEX_CACHE = {} +_ENTRIES_INDEX_TTL = 600 # seconds; a newly added index is used within this + + +def _entries_indexes(rdb_session): + """(index names, first columns) of the entries table, as the database + has them: production's indexes differ from the model's (no index on + name, ix_entries_file_id only through tools/migrate_prod_db.sql).""" + # Reflect on the session's own connection: on the engine, SQLAlchemy + # uses a connection that closes after its first result, and the app's + # engine_connect listener (SET NAMES, app.py) uses up that result, so + # the reflection then fails with "This Connection is closed". + connection = rdb_session.connection() + engine = connection.engine + cached = _ENTRIES_INDEX_CACHE.get(id(engine)) + if cached and time.time() - cached[0] < _ENTRIES_INDEX_TTL: + return cached[1] + indexes = inspect(connection).get_indexes('entries') + found = (set(i['name'] for i in indexes), + set(i['column_names'][0] for i in indexes if i['column_names'])) + _ENTRIES_INDEX_CACHE[id(engine)] = (time.time(), found) + return found + + class RoundSource(Base): __tablename__ = 'round_sources' @@ -1212,6 +1275,18 @@ def get_entry_name_map(self, filenames): ret[name] = entry return ret + def get_entry_file_id_map(self, file_ids): + # Keep len(file_ids) below 200: see FILE_ID_LOOKUP_CHUNK_SIZE. + entries = (self.query(Entry) + .filter(Entry.file_id.in_(file_ids)) + .order_by(Entry.id) + .all()) + ret = {} + for entry in entries: + # with duplicates, the oldest row wins, as a stable choice + ret.setdefault(entry.file_id, entry) + return ret + def get_grouped_flags(self, round_id): flagged_entries = (self.query(RoundEntry) .filter_by(round_id=round_id) @@ -1675,6 +1750,9 @@ def add_entries(self, rnd, entries): deduped.append(e) entries = deduped + if _use_file_id_lookup(rnd, entries): + return self._add_entries_by_file_id(entries) + entry_chunks = chunked(entries, IMPORT_CHUNK_SIZE) ret = [] new_entry_count = 0 @@ -1699,6 +1777,47 @@ def add_entries(self, rnd, entries): return ret, new_entry_count + def _add_entries_by_file_id(self, entries): + """add_entries for imports where every entry has a file_id (see + _use_file_id_lookup): existing rows are found, and new rows read + back, by file_id instead of by name.""" + seen = set() + deduped = [] + for e in entries: + if e.file_id not in seen: + seen.add(e.file_id) + deduped.append(e) + + # Files imported before the cutoff have no file_id; find them by name + # where that is cheap (an index on name), so they keep their row. + # Production has no such index: there they get a second row. + name_fallback = 'name' in _entries_indexes(self.rdb_session)[1] + + ret = [] + new_entry_count = 0 + for entry_chunk in chunked(deduped, FILE_ID_LOOKUP_CHUNK_SIZE): + db_entries = self.get_entry_file_id_map( + [e.file_id for e in entry_chunk]) + missing = [e for e in entry_chunk if e.file_id not in db_entries] + if missing and name_fallback: + by_name = self.get_entry_name_map( + [to_unicode(e.name) for e in missing]) + for e in missing: + row = by_name.get(to_unicode(e.name)) + if row is not None and row.file_id is None: + db_entries[e.file_id] = row + new_rows = [_entry_insert_row(e) for e in entry_chunk + if e.file_id not in db_entries] + if new_rows: + self.rdb_session.execute(entries_t.insert(), new_rows) + new_entry_count += len(new_rows) + db_entries.update( + self.get_entry_file_id_map([r['file_id'] for r in new_rows])) + for entry in entry_chunk: + ret.append(db_entries[entry.file_id]) + + return ret, new_entry_count + def add_round_entries(self, round_id, entries, method, params): rnd = self.user_dao.get_round(round_id) if rnd.status != PAUSED_STATUS: diff --git a/montage/tests/test_import_entries.py b/montage/tests/test_import_entries.py index 6569915a..d9a4f22f 100644 --- a/montage/tests/test_import_entries.py +++ b/montage/tests/test_import_entries.py @@ -10,6 +10,7 @@ """ import datetime import json +import zlib from unittest.mock import patch import pytest @@ -161,15 +162,16 @@ def mock_category(mock_external_apis, file_infos): status=200) -def new_round(api_client, name, config=None, quorum=None): +def new_round(api_client, name, config=None, quorum=None, open_date=OPEN_DATE, + close_date='2016-01-01T00:00:00'): """A paused round in a fresh campaign; returns the round id.""" series_id = api_client.fetch('get default series', '/series')['data'][0]['id'] campaign_id = api_client.fetch( 'organizer: create campaign', '/admin/add_campaign', {'name': name, 'coordinators': [COORD], - 'open_date': OPEN_DATE.isoformat(), - 'close_date': '2016-01-01T00:00:00', + 'open_date': open_date.isoformat(), + 'close_date': close_date, 'url': 'http://hatnote.com', 'series_id': series_id}, as_user=COORD)['data']['id'] @@ -476,3 +478,174 @@ def test_mysql_driver_batches_the_inserts(): ['entry_id', 'round_id', 'round_source_id'])): sql = str(table.insert().compile(dialect=dialect, column_keys=keys)) assert RE_INSERT_VALUES.match(sql), sql + + +# --------------------------------------------------------------------------- +# Lookup by file_id for campaigns after the cutoff (#513) +# --------------------------------------------------------------------------- + +AFTER_CUTOFF = rdb.FILE_ID_LOOKUP_CUTOFF + datetime.timedelta(days=30) + + +def with_file_ids(infos): + """The same files, all with a file_id (as wikireplica imports give); + a missing one is derived from the name, so it is stable and unique.""" + return [dict(info, file_id=info['file_id'] + or 10 ** 7 + zlib.crc32(info['img_name'].encode('utf8'))) + for info in infos] + + +def new_round_after_cutoff(api_client, name): + return new_round(api_client, name, open_date=AFTER_CUTOFF, + close_date='2026-12-31T00:00:00') + + +def _no_name_lookup(*a, **kw): + raise AssertionError('looked up entries by name') + + +def _no_file_id_lookup(*a, **kw): + raise AssertionError('looked up entries by file_id') + + +def test_file_id_lookup_after_cutoff(montage_app, coord_client, + mock_external_apis, monkeypatch): + first = with_file_ids(make_file_infos('fid', IMPORT_CHUNK_SIZE + 50)) + more = with_file_ids(make_file_infos('fid', 30, start=500)) + round_id = new_round_after_cutoff(coord_client, 'fid') + # production's indexes on entries: file_id yes, name no + monkeypatch.setattr(rdb, '_entries_indexes', lambda session: ( + {'PRIMARY', 'ix_entries_file_id'}, {'id', 'file_id'})) + monkeypatch.setattr(CoordinatorDAO, 'get_entry_name_map', _no_name_lookup) + + mock_category(mock_external_apis, first) + data = import_category(coord_client, round_id)['data'] + assert data['new_round_entry_count'] == len(first) + ids_before = entry_ids(montage_app, 'fid') + assert len(ids_before) == len(first) + + mock_category(mock_external_apis, first + more) + data = import_category(coord_client, round_id)['data'] + assert data['new_round_entry_count'] == 30 + ids_after = entry_ids(montage_app, 'fid') + assert len(ids_after) == len(first) + 30 + assert {k: ids_after[k] for k in ids_before} == ids_before + + # a second campaign after the cutoff reuses the same rows + other = new_round_after_cutoff(coord_client, 'fid other') + data = import_category(coord_client, other)['data'] + assert data['new_round_entry_count'] == len(first) + 30 + assert entry_ids(montage_app, 'fid') == ids_after + + +def test_name_lookup_before_cutoff(montage_app, coord_client, + mock_external_apis, monkeypatch): + infos = with_file_ids(make_file_infos('old', 40)) + round_id = new_round(coord_client, 'old') # opens in 2015 + monkeypatch.setattr(CoordinatorDAO, 'get_entry_file_id_map', + _no_file_id_lookup) + mock_category(mock_external_apis, infos) + import_category(coord_client, round_id) + assert len(entry_ids(montage_app, 'old')) == 40 + + +def test_name_lookup_when_a_file_id_is_missing(montage_app, coord_client, + mock_external_apis, + monkeypatch): + infos = make_file_infos('nofid', 40) # some have file_id None + assert any(info['file_id'] is None for info in infos) + round_id = new_round_after_cutoff(coord_client, 'nofid') + monkeypatch.setattr(CoordinatorDAO, 'get_entry_file_id_map', + _no_file_id_lookup) + mock_category(mock_external_apis, infos) + import_category(coord_client, round_id) + assert len(entry_ids(montage_app, 'nofid')) == 40 + + +def test_file_id_lookup_keeps_the_row_of_a_renamed_file(montage_app, + coord_client, + mock_external_apis): + infos = with_file_ids(make_file_infos('ren', 3)) + round_a = new_round_after_cutoff(coord_client, 'ren a') + mock_category(mock_external_apis, infos) + import_category(coord_client, round_a) + ids = entry_ids(montage_app, 'ren') + + renamed = [dict(infos[0], img_name='ren_renamed.jpg')] + infos[1:] + round_b = new_round_after_cutoff(coord_client, 'ren b') + mock_category(mock_external_apis, renamed) + data = import_category(coord_client, round_b)['data'] + assert data['new_round_entry_count'] == 3 + assert entry_ids(montage_app, 'ren') == ids # no row for the new name + + +def test_file_id_lookup_keeps_the_row_of_a_file_without_file_id( + montage_app, coord_client, mock_external_apis): + # imported before the cutoff, so stored without a file_id + infos = with_file_ids(make_file_infos('pre', 5)) + old_round = new_round(coord_client, 'pre old') + mock_category(mock_external_apis, + [dict(info, file_id=None) for info in infos]) + import_category(coord_client, old_round) + ids = entry_ids(montage_app, 'pre') + + # the same files after the cutoff, now with a file_id: where entries.name + # is indexed (here unique, as on beta) they keep their rows + new = new_round_after_cutoff(coord_client, 'pre new') + mock_category(mock_external_apis, infos) + data = import_category(coord_client, new)['data'] + assert data['new_round_entry_count'] == 5 + assert entry_ids(montage_app, 'pre') == ids + + +def test_entries_indexes_works_with_the_apps_connect_listener(): + """The app runs SET NAMES on every new connection (app.py); reflecting + on the engine then hit a closed connection on montage-dev.""" + from sqlalchemy import create_engine, event + from sqlalchemy.orm import sessionmaker + engine = create_engine('sqlite://') + rdb.Base.metadata.create_all(engine) + + def like_set_names(connection, branch): + connection.execute('PRAGMA foreign_keys = ON') # no rows, like SET NAMES + + event.listen(engine, 'engine_connect', like_set_names) + session = sessionmaker(bind=engine)() + session.query(Entry.id).first() + rdb._ENTRIES_INDEX_CACHE.clear() + names, first_columns = rdb._entries_indexes(session) + assert 'name' in first_columns + session.query(Entry.id).first() # the session is still usable + + +def test_entries_indexes_reads_the_database(): + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + engine = create_engine('sqlite://') + rdb.Base.metadata.create_all(engine) + names, first_columns = rdb._entries_indexes(sessionmaker(bind=engine)()) + assert 'name' in first_columns # the model's unique index + assert 'ix_entries_file_id' not in names # only migrate_prod_db.sql adds it + + +def test_file_id_lookups_stay_below_the_index_dive_limit( + montage_app, coord_client, mock_external_apis, monkeypatch): + """MariaDB stops using ix_entries_file_id for IN lists of 200 values or + more (eq_range_index_dive_limit); see rdb.FILE_ID_LOOKUP_CHUNK_SIZE.""" + assert rdb.FILE_ID_LOOKUP_CHUNK_SIZE < 200 + sizes = [] + real_lookup = CoordinatorDAO.get_entry_file_id_map + + def recording_lookup(self, file_ids): + sizes.append(len(file_ids)) + return real_lookup(self, file_ids) + + monkeypatch.setattr(CoordinatorDAO, 'get_entry_file_id_map', + recording_lookup) + infos = with_file_ids(make_file_infos('dive', 450)) + round_id = new_round_after_cutoff(coord_client, 'dive') + mock_category(mock_external_apis, infos) + import_category(coord_client, round_id) + + assert len(entry_ids(montage_app, 'dive')) == 450 + assert sizes and max(sizes) <= rdb.FILE_ID_LOOKUP_CHUNK_SIZE