From 251d09e04d3d385de2844ea3ca7776f4e4368969 Mon Sep 17 00:00:00 2001 From: lgelauff <32471480+lgelauff@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:57:27 +0000 Subject: [PATCH 1/2] perf(import): look up existing entries by file_id, in batches of 199 The import looks up existing entries by name in batches of 200, twice per batch (lookup, and read-back after the multi-row insert). Two things make that slow on production (hatnote/montage#513, #650): - Production has no index on entries.name (lost when the tables were converted to utf8mb4_unicode_ci / Compact on 2025-07-01), so every batch reads the whole 2M-row table. On a copy of production (#652) a ~6,000-file category import was killed after 31 s, nothing saved. - MariaDB's eq_range_index_dive_limit is 200: from 200 values in an IN (...) list on, the optimizer estimates from index statistics instead of counting in the index. Measured on production (read-only, 2026-10-05) for newest, oldest, random and new file_ids: 50, 150 and 199 ids always use ix_entries_file_id (1-7 ms, also with 0 hits), 200, 201 and 400 ids always read the whole table (0.56 s warm, 9.4 s cold). Change: for campaigns opened on or after 2026-06-01, when every file in the import has a file_id (wikireplica imports; CSV imports have none), add_entries finds existing rows, and reads new rows back, by file_id in batches of 199 (FILE_ID_LOOKUP_CHUNK_SIZE). No index hint is needed; FORCE INDEX is documented next to the constant as a fallback. On production the last category-imported entry without a file_id dates from 2026-05-30 13:21 and the first with one from 2026-05-31 20:52, so every file new to Montage after the cutoff has its file_id. Files not found by file_id are looked up by name only where entries.name is indexed (beta, fresh installs, production once #650's index is restored); a row found that way is reused if it has no file_id, so older files keep their row and the import does not fail on beta's unique name index. Without a name index (production today) that lookup is skipped; such files get a second row. The indexes are read from the database and cached for ten minutes. A renamed file (same file_id) keeps its existing row. Other imports are unchanged; no schema change. Tests: no name lookup after the cutoff with production's indexes, and re-imports keep row ids; name lookup before the cutoff and without a file_id; a renamed file keeps its row; a file without file_id keeps its row where name is indexed; every file_id lookup stays below 200 values. Each fails without its part of the change. Co-Authored-By: Claude Opus 5.5 --- montage/rdb.py | 115 +++++++++++++++++++ montage/tests/test_import_entries.py | 159 ++++++++++++++++++++++++++- 2 files changed, 271 insertions(+), 3 deletions(-) diff --git a/montage/rdb.py b/montage/rdb.py index db45356f..56c7761f 100644 --- a/montage/rdb.py +++ b/montage/rdb.py @@ -654,6 +654,65 @@ 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).""" + engine = rdb_session.get_bind() + engine = getattr(engine, 'engine', engine) + cached = _ENTRIES_INDEX_CACHE.get(id(engine)) + if cached and time.time() - cached[0] < _ENTRIES_INDEX_TTL: + return cached[1] + indexes = inspect(engine).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 +1271,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 +1746,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 +1773,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..f34f6466 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,154 @@ 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_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 From 387a7490efc46d07b602d0538dc20b713dd43d9c Mon Sep 17 00:00:00 2001 From: lgelauff <32471480+lgelauff@users.noreply.github.com> Date: Mon, 5 Oct 2026 10:19:03 +0000 Subject: [PATCH 2/2] fix(import): read entries' indexes on the session's connection Found on montage-dev: a 17,000-file import failed after 4 s with "ResourceClosedError: This Connection is closed [SQL: 'SHOW CREATE TABLE entries']" from _entries_indexes. inspect(engine) reflects through connectionless execution, on a connection that closes itself once its first result is exhausted. The app's engine_connect listener (set_mysql_session_charset_and_collation) runs SET NAMES on every new connection; that result has no rows, so it is exhausted at once and the connection closes before the reflection query runs. Reflecting on rdb_session.connection() uses the request's own connection instead (no new connection, same transaction). Reproduced and verified on MariaDB with the app's engine and listeners (old: fails with the same error; new: indexes read, session still usable). Test: an engine_connect listener that runs a statement without rows (like SET NAMES) no longer breaks _entries_indexes; fails with the old code with the same ResourceClosedError. Co-Authored-By: Claude Opus 5.5 --- montage/rdb.py | 10 +++++++--- montage/tests/test_import_entries.py | 20 ++++++++++++++++++++ 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/montage/rdb.py b/montage/rdb.py index 56c7761f..f0894377 100644 --- a/montage/rdb.py +++ b/montage/rdb.py @@ -701,12 +701,16 @@ 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).""" - engine = rdb_session.get_bind() - engine = getattr(engine, 'engine', engine) + # 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(engine).get_indexes('entries') + 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) diff --git a/montage/tests/test_import_entries.py b/montage/tests/test_import_entries.py index f34f6466..d9a4f22f 100644 --- a/montage/tests/test_import_entries.py +++ b/montage/tests/test_import_entries.py @@ -598,6 +598,26 @@ def test_file_id_lookup_keeps_the_row_of_a_file_without_file_id( 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