From d450a7e50444556ca88c89f34edbfd743fd7f6cb Mon Sep 17 00:00:00 2001 From: Alam-2U Date: Thu, 25 Jun 2026 05:46:41 +0000 Subject: [PATCH 1/4] fix: optimize forum migration to eliminate N+1 queries and improve scalability --- forum/__init__.py | 2 +- ...um_migrate_course_from_mongodb_to_mysql.py | 141 +- forum/migration_helpers.py | 1244 ++++++++++++----- 3 files changed, 989 insertions(+), 398 deletions(-) diff --git a/forum/__init__.py b/forum/__init__.py index 17007c27..186652c8 100644 --- a/forum/__init__.py +++ b/forum/__init__.py @@ -2,4 +2,4 @@ Openedx forum app. """ -__version__ = "0.6.7" +__version__ = "0.6.8" diff --git a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py index 298d47a4..48b82fb0 100644 --- a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py +++ b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py @@ -1,11 +1,15 @@ """Migration command for courses from mongodb to mysql.""" +import time +from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any from django.core.management.base import BaseCommand from django.core.management.base import CommandParser +from django.db import connections from forum.migration_helpers import ( + BATCH_SIZE, enable_mysql_backend_for_course, get_all_course_ids, migrate_content, @@ -14,6 +18,52 @@ ) from forum.mongo import get_database +# --------------------------------------------------------------------------- +# Default parallelism. +# 1 = single-threaded (safe for SQLite / test environments, default). +# N = N threads, each with its own Django DB connection; requires a +# thread-safe database backend (MySQL, PostgreSQL) for N > 1. +# --------------------------------------------------------------------------- +DEFAULT_WORKERS = 1 + + +def _migrate_one_course( + course_id: str, + create_waffle_flags: bool, +) -> tuple[str, float | None, str | None]: + """ + Migrate a single course in the calling thread. + + Returns ``(course_id, elapsed_seconds, error_message_or_None)``. + Each thread gets its own MongoDB client and Django DB connection. + """ + # Ensure this thread does not reuse a connection inherited from the + # main thread (Django creates a new one on first access per thread). + connections.close_all() + + # Django stores DB connections in thread-local storage, so each worker + # thread automatically gets its own connection on first access. + # Explicitly close any connection inherited from the spawning thread + # so the worker starts with a clean slate. + connections.close_all() + + db = get_database() + t0 = time.monotonic() + try: + migrate_users(db, course_id) + migrate_content(db, course_id) + migrate_read_states(db, course_id) + if create_waffle_flags: + enable_mysql_backend_for_course(course_id) + elapsed = time.monotonic() - t0 + return course_id, elapsed, None + except Exception as exc: # pylint: disable=broad-except + elapsed = time.monotonic() - t0 + return course_id, elapsed, str(exc) + finally: + # Release the DB connection so the pool slot is returned promptly. + connections.close_all() + class Command(BaseCommand): """Migration command for courses from mongodb to mysql.""" @@ -28,6 +78,27 @@ def add_arguments(self, parser: CommandParser) -> None: action="store_true", help="Skip course waffle flag creation", ) + parser.add_argument( + "-w", + "--workers", + type=int, + default=DEFAULT_WORKERS, + metavar="N", + help=( + f"Number of parallel worker threads (default: {DEFAULT_WORKERS}). " + "Each worker processes one course at a time and uses its own " + "database connection. Set > 1 only with a thread-safe DB " + "backend (MySQL, PostgreSQL). Example: --workers 8" + ), + ) + parser.add_argument( + "-b", + "--batch-size", + type=int, + default=BATCH_SIZE, + metavar="N", + help=f"Bulk-operation batch size (default: {BATCH_SIZE}).", + ) parser.add_argument( "courses", nargs="+", type=str, help="List of course IDs or `all`" ) @@ -37,19 +108,71 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: db = get_database() create_waffle_flags = not options["no_toggle"] + workers: int = int(options["workers"]) # type: ignore[arg-type] + + # Override module-level BATCH_SIZE when caller passes --batch-size. + import forum.migration_helpers as _mh + _mh.BATCH_SIZE = int(options["batch_size"]) # type: ignore[arg-type] + course_ids = list(options["courses"]) if "all" in course_ids: course_ids = get_all_course_ids(db) - for course_id in course_ids: - self.stdout.write(f"Migrating data for course: {course_id}") - migrate_users(db, course_id) - migrate_content(db, course_id) - migrate_read_states(db, course_id) - if create_waffle_flags: - enable_mysql_backend_for_course(course_id) - self.stdout.write( - f"Enabled mysql backend waffle flag for course {course_id}." + total = len(course_ids) + self.stdout.write( + f"Migrating {total} course(s) with {workers} parallel worker(s) " + f"(batch_size={_mh.BATCH_SIZE})." + ) + + failed: list[tuple[str, str]] = [] + completed = 0 + + if workers == 1: + # Single-threaded path: simpler, no executor overhead. + for course_id in course_ids: + cid, elapsed, err = _migrate_one_course(course_id, create_waffle_flags) + completed += 1 + if err: + self.stderr.write( + self.style.ERROR( + f"[{completed}/{total}] FAILED {cid} after {elapsed:.1f}s: {err}" + ) + ) + failed.append((cid, err)) + else: + self.stdout.write( + f"[{completed}/{total}] OK {cid} ({elapsed:.1f}s)" + ) + else: + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = { + pool.submit(_migrate_one_course, cid, create_waffle_flags): cid + for cid in course_ids + } + for future in as_completed(futures): + cid, elapsed, err = future.result() + completed += 1 + if err: + self.stderr.write( + self.style.ERROR( + f"[{completed}/{total}] FAILED {cid} after " + f"{elapsed:.1f}s: {err}" + ) + ) + failed.append((cid, err)) + else: + self.stdout.write( + f"[{completed}/{total}] OK {cid} ({elapsed:.1f}s)" + ) + + if failed: + self.stderr.write( + self.style.ERROR( + f"\n{len(failed)} course(s) failed migration:" ) + ) + for cid, err in failed: + self.stderr.write(self.style.ERROR(f" {cid}: {err}")) + raise SystemExit(1) self.stdout.write(self.style.SUCCESS("Data migration completed successfully")) diff --git a/forum/migration_helpers.py b/forum/migration_helpers.py index 7cad1a7d..eb64cae2 100644 --- a/forum/migration_helpers.py +++ b/forum/migration_helpers.py @@ -5,6 +5,7 @@ from typing import Any from django.contrib.auth.models import User # pylint: disable=E5142 +from django.contrib.contenttypes.models import ContentType from django.core.management.base import OutputWrapper from django.utils import timezone from pymongo.collection import Collection @@ -28,6 +29,12 @@ logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Batch size for all bulk_create / bulk_update operations. +# Tune this to balance memory use vs round-trip overhead. +# --------------------------------------------------------------------------- +BATCH_SIZE = 500 + def get_user_or_none(user_id: Any) -> User | None: """Get a user by ID or return None if not found.""" @@ -59,448 +66,909 @@ def get_all_course_ids(db: Database[dict[str, Any]]) -> list[str]: return db.contents.distinct("course_id") +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _to_int_id(uid: Any) -> int | None: + """Convert a user ID value to int; return None on failure.""" + try: + return int(uid) + except (ValueError, TypeError): + return None + + +def _collect_all_user_ids(contents: list[dict[str, Any]]) -> set[int]: + """Return the set of all integer user IDs referenced in *contents*.""" + ids: set[int] = set() + for c in contents: + for field in ("author_id", "deleted_by", "closed_by_id"): + uid = _to_int_id(c.get(field)) + if uid is not None: + ids.add(uid) + for vote_type in ("up", "down"): + for uid_str in c.get("votes", {}).get(vote_type, []): + uid = _to_int_id(uid_str) + if uid is not None: + ids.add(uid) + for uid_str in c.get("abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None: + ids.add(uid) + for uid_str in c.get("historical_abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None: + ids.add(uid) + for edit in c.get("edit_history", []): + uid = _to_int_id(edit.get("author_id")) + if uid is not None: + ids.add(uid) + return ids + + +def _build_user_cache(user_ids: set[int]) -> dict[int, User]: + """Fetch all requested users in one query; return a {pk: User} mapping.""" + if not user_ids: + return {} + return {u.pk: u for u in User.objects.filter(pk__in=user_ids)} + + +# --------------------------------------------------------------------------- +# migrate_users (batch-optimised) +# --------------------------------------------------------------------------- + def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: - """Migrate users from MongoDB to MySQL.""" - users = db.users.find({"course_stats.course_id": course_id}) - for user_data in users: - user = get_user_or_none(user_data["_id"]) - if not user: - continue + """ + Migrate users from MongoDB to MySQL. - _, _ = ForumUser.objects.get_or_create( - user=user, - defaults={"default_sort_key": user_data.get("default_sort_key", "date")}, + Uses bulk_create / bulk_update instead of per-row get_or_create calls, + reducing the number of SQL round-trips from O(N) to O(1). + """ + all_mongo_users = list(db.users.find({"course_stats.course_id": course_id})) + if not all_mongo_users: + return + + # Build a numeric-uid → mongo-doc map, dropping unparsable IDs. + uid_map: dict[int, dict[str, Any]] = {} + for u in all_mongo_users: + uid = _to_int_id(u["_id"]) + if uid is not None: + uid_map[uid] = u + + if not uid_map: + return + + # --- Single bulk fetch of all Django users for this course --- + django_users = {u.pk: u for u in User.objects.filter(pk__in=uid_map.keys())} + + # --- ForumUser: create missing rows in one shot --- + existing_fu_ids = set( + ForumUser.objects.filter(user_id__in=django_users.keys()).values_list( + "user_id", flat=True + ) + ) + new_forum_users = [ + ForumUser( + user_id=uid, + default_sort_key=uid_map[uid].get("default_sort_key", "date"), + ) + for uid in django_users + if uid not in existing_fu_ids + ] + if new_forum_users: + ForumUser.objects.bulk_create( + new_forum_users, batch_size=BATCH_SIZE, ignore_conflicts=True ) - for stat in user_data.get("course_stats", []): + # --- CourseStat: bulk create new / bulk update existing --- + existing_stats: dict[int, CourseStat] = { + cs.user_id: cs + for cs in CourseStat.objects.filter( + user_id__in=django_users.keys(), course_id=course_id + ) + } + stat_update_fields = [ + "active_flags", "inactive_flags", "threads", "responses", "replies", + "deleted_threads", "deleted_responses", "deleted_replies", "last_activity_at", + ] + stats_to_create: list[CourseStat] = [] + stats_to_update: list[CourseStat] = [] + + for uid, mongo_user in uid_map.items(): + if uid not in django_users: + continue + for stat in mongo_user.get("course_stats", []): + if stat.get("course_id") != course_id: + continue last_activity_at = parse_mongo_datetime(stat.get("last_activity_at")) - if stat["course_id"] == course_id: - CourseStat.objects.update_or_create( - user=user, - course_id=course_id, - defaults={ - "active_flags": stat.get("active_flags", 0), - "inactive_flags": stat.get("inactive_flags", 0), - "threads": stat.get("threads", 0), - "responses": stat.get("responses", 0), - "replies": stat.get("replies", 0), - "deleted_threads": stat.get("deleted_threads", 0), - "deleted_responses": stat.get("deleted_responses", 0), - "deleted_replies": stat.get("deleted_replies", 0), - "last_activity_at": last_activity_at, - }, + if uid in existing_stats: + cs = existing_stats[uid] + cs.active_flags = stat.get("active_flags", 0) + cs.inactive_flags = stat.get("inactive_flags", 0) + cs.threads = stat.get("threads", 0) + cs.responses = stat.get("responses", 0) + cs.replies = stat.get("replies", 0) + cs.deleted_threads = stat.get("deleted_threads", 0) + cs.deleted_responses = stat.get("deleted_responses", 0) + cs.deleted_replies = stat.get("deleted_replies", 0) + cs.last_activity_at = last_activity_at + stats_to_update.append(cs) + else: + stats_to_create.append( + CourseStat( + user_id=uid, + course_id=course_id, + active_flags=stat.get("active_flags", 0), + inactive_flags=stat.get("inactive_flags", 0), + threads=stat.get("threads", 0), + responses=stat.get("responses", 0), + replies=stat.get("replies", 0), + deleted_threads=stat.get("deleted_threads", 0), + deleted_responses=stat.get("deleted_responses", 0), + deleted_replies=stat.get("deleted_replies", 0), + last_activity_at=last_activity_at, + ) ) + if stats_to_create: + CourseStat.objects.bulk_create( + stats_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if stats_to_update: + CourseStat.objects.bulk_update( + stats_to_update, stat_update_fields, batch_size=BATCH_SIZE + ) + + +# --------------------------------------------------------------------------- +# migrate_content (batch-optimised) +# --------------------------------------------------------------------------- def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: - """Migrate content from MongoDB to MySQL.""" - contents = db.contents.find({"course_id": course_id}).sort("created_at") - for content in contents: - if content["_type"] == "CommentThread": - create_or_update_thread(content) - elif content["_type"] == "Comment": - create_or_update_comment(content) - - migrate_subscriptions(db, content["_id"]) - - -def _resolve_thread_actors( - thread_data: dict[str, Any], -) -> tuple["User | None", "User | None"]: - """Return (deleted_by, closed_by) users resolved from thread_data.""" - deleted_by = ( - get_user_or_none(thread_data["deleted_by"]) - if thread_data.get("deleted_by") - else None + """ + Migrate threads and comments from MongoDB to MySQL. + + Strategy: + 1. Fetch ALL content for the course in a single MongoDB query. + 2. Build a user cache with a single Django query (eliminates the + biggest N+1 problem: per-voter/flagger/editor user lookups). + 3. Pre-fetch all existing MongoContent mappings in one query. + 4. Bulk-create new threads, then top-level comments, then child + comments (preserving the FK ordering requirement). + 5. Bulk-create votes, edit history, and abuse flaggers. + 6. Issue a SINGLE MongoDB ``$in`` query for all subscriptions + instead of one query per content item. + """ + contents = list( + db.contents.find({"course_id": course_id}).sort("created_at", 1) ) - closed_by = ( - get_user_or_none(thread_data["closed_by_id"]) - if thread_data.get("closed_by_id") - else None + if not contents: + return + + all_ids_str = [str(c["_id"]) for c in contents] + + # Pre-fetch all existing MongoContent rows (one query). + mongo_cache: dict[str, MongoContent] = { + mc.mongo_id: mc + for mc in MongoContent.objects.filter(mongo_id__in=all_ids_str) + } + + # Collect every user ID referenced anywhere; fetch them all at once. + user_cache = _build_user_cache(_collect_all_user_ids(contents)) + + # ContentType objects are cached by Django's framework after the first call. + thread_ct = ContentType.objects.get_for_model(CommentThread) + comment_ct = ContentType.objects.get_for_model(Comment) + + threads_data = [c for c in contents if c["_type"] == "CommentThread"] + comments_data = [c for c in contents if c["_type"] == "Comment"] + + # --- Threads --- + _bulk_migrate_threads(threads_data, mongo_cache, user_cache, thread_ct) + _refresh_mongo_cache(mongo_cache, [str(t["_id"]) for t in threads_data]) + + # --- Comments: top-level first, then children (parent must exist first) --- + top_level = [ + c for c in comments_data + if not c.get("parent_id") or str(c.get("parent_id")) == "None" + ] + child_comments = [ + c for c in comments_data + if c.get("parent_id") and str(c.get("parent_id")) != "None" + ] + _bulk_migrate_comments(top_level, mongo_cache, user_cache, comment_ct) + _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in top_level]) + _bulk_migrate_comments(child_comments, mongo_cache, user_cache, comment_ct) + _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in child_comments]) + + # --- Metadata: votes, edit history, flaggers --- + _bulk_migrate_votes(contents, mongo_cache, user_cache, thread_ct, comment_ct) + _bulk_migrate_edit_history(contents, mongo_cache, user_cache, thread_ct, comment_ct) + _bulk_migrate_abuse_flaggers(contents, mongo_cache, user_cache, thread_ct, comment_ct) + + # --- Subscriptions: ONE MongoDB query for the entire course --- + _bulk_migrate_subscriptions(db, all_ids_str, mongo_cache, user_cache) + + +def _refresh_mongo_cache( + cache: dict[str, MongoContent], mongo_ids: list[str] +) -> None: + """Update *cache* with freshly queried MongoContent rows for *mongo_ids*.""" + if not mongo_ids: + return + cache.update( + {mc.mongo_id: mc for mc in MongoContent.objects.filter(mongo_id__in=mongo_ids)} ) - return deleted_by, closed_by -def create_or_update_thread(thread_data: dict[str, Any]) -> None: - """Create or update a thread.""" - author = get_user_or_none(thread_data["author_id"]) - if not author: +def _bulk_migrate_threads( + threads_data: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, +) -> None: + """Bulk-create new threads and bulk-update existing ones.""" + if not threads_data: return - # Preserve author_username from MongoDB (historical username) - author_username = thread_data.get("author_username") + new_data = [ + t for t in threads_data + if not (mongo_cache.get(str(t["_id"])) and + mongo_cache[str(t["_id"])].content_object_id) + ] + existing_data = [ + t for t in threads_data + if mongo_cache.get(str(t["_id"])) and + mongo_cache[str(t["_id"])].content_object_id + ] + + # --- Create new threads in bulk --- + if new_data: + pairs: list[tuple[str, CommentThread]] = [] + for t in new_data: + author = user_cache.get(_to_int_id(t.get("author_id"))) # type: ignore[arg-type] + if not author: + continue + author_username = ( + t.get("author_username") + or t.get("retired_username") + or author.username + ) + deleted_by = ( + user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] + if t.get("deleted_by") else None + ) + closed_by = ( + user_cache.get(_to_int_id(t.get("closed_by_id"))) # type: ignore[arg-type] + if t.get("closed_by_id") else None + ) + pairs.append(( + str(t["_id"]), + CommentThread( + author=author, + author_username=author_username, + retired_username=t.get("retired_username"), + course_id=t["course_id"], + title=get_trunc_title(t.get("title", "")), + body=t["body"], + thread_type=t.get("thread_type", "discussion"), + context=t.get("context", "course"), + anonymous=t.get("anonymous", False), + anonymous_to_peers=t.get("anonymous_to_peers", False), + closed=t.get("closed", False), + closed_by=closed_by, + close_reason_code=t.get("close_reason_code"), + pinned=t.get("pinned", False), + created_at=parse_mongo_datetime(t["created_at"]), + updated_at=parse_mongo_datetime(t["updated_at"]), + last_activity_at=parse_mongo_datetime(t["last_activity_at"]), + commentable_id=t.get("commentable_id"), + is_spam=t.get("is_spam", False), + is_deleted=t.get("is_deleted", False), + deleted_at=parse_mongo_datetime(t.get("deleted_at")), + deleted_by=deleted_by, + visible=t.get("visible", True), + ), + )) + + if pairs: + mongo_ids, objs = zip(*pairs) + # Django 4.1+ returns PKs from bulk_create on MySQL 8.0.19+. + created = CommentThread.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) + mc_rows = [ + MongoContent( + mongo_id=mid, + content_type=thread_ct, + content_object_id=thread.pk, + ) + for mid, thread in zip(mongo_ids, created) + if thread.pk + ] + if mc_rows: + MongoContent.objects.bulk_create( + mc_rows, batch_size=BATCH_SIZE, ignore_conflicts=True + ) - # Preserve retired_username from MongoDB - retired_username = thread_data.get("retired_username") + # --- Update existing threads in bulk --- + if existing_data: + pks = [ + mongo_cache[str(t["_id"])].content_object_id + for t in existing_data + if str(t["_id"]) in mongo_cache + ] + thread_pk_map = { + th.pk: th + for th in CommentThread.objects.filter(pk__in=pks) + } + thread_update_fields = [ + "title", "body", "thread_type", "context", "anonymous", + "anonymous_to_peers", "closed", "closed_by", "close_reason_code", + "pinned", "updated_at", "last_activity_at", "commentable_id", + "is_spam", "is_deleted", "deleted_at", "deleted_by", "visible", + ] + to_update: list[CommentThread] = [] + for t in existing_data: + mc = mongo_cache.get(str(t["_id"])) + if not mc: + continue + thread = thread_pk_map.get(mc.content_object_id) + if not thread: + continue + deleted_by = ( + user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] + if t.get("deleted_by") else None + ) + closed_by = ( + user_cache.get(_to_int_id(t.get("closed_by_id"))) # type: ignore[arg-type] + if t.get("closed_by_id") else None + ) + thread.title = get_trunc_title(t.get("title", "")) + thread.body = t["body"] + thread.thread_type = t.get("thread_type", "discussion") + thread.context = t.get("context", "course") + thread.anonymous = t.get("anonymous", False) + thread.anonymous_to_peers = t.get("anonymous_to_peers", False) + thread.closed = t.get("closed", False) + thread.closed_by = closed_by # type: ignore[assignment] + thread.close_reason_code = t.get("close_reason_code") + thread.pinned = t.get("pinned", False) + thread.updated_at = parse_mongo_datetime(t["updated_at"]) # type: ignore[assignment] + thread.last_activity_at = parse_mongo_datetime(t["last_activity_at"]) + thread.commentable_id = t.get("commentable_id") # type: ignore[assignment] + thread.is_spam = t.get("is_spam", False) + thread.is_deleted = t.get("is_deleted", False) + thread.deleted_at = parse_mongo_datetime(t.get("deleted_at")) + thread.deleted_by = deleted_by # type: ignore[assignment] + thread.visible = t.get("visible", True) + to_update.append(thread) + if to_update: + CommentThread.objects.bulk_update( + to_update, thread_update_fields, batch_size=BATCH_SIZE + ) - # If author_username is not provided, use retired_username or fallback to current username - if not author_username: - if retired_username: - author_username = retired_username - else: - author_username = author.username - mongo_thread_id = str(thread_data["_id"]) - mongo_content, _ = MongoContent.objects.get_or_create( - mongo_id=mongo_thread_id, - ) - if not mongo_content.content_object_id: - deleted_by, closed_by = _resolve_thread_actors(thread_data) - - thread = CommentThread.objects.create( - author=author, - author_username=author_username, - retired_username=retired_username, - course_id=thread_data["course_id"], - title=get_trunc_title(thread_data.get("title", "")), - body=thread_data["body"], - thread_type=thread_data.get("thread_type", "discussion"), - context=thread_data.get("context", "course"), - anonymous=thread_data.get("anonymous", False), - anonymous_to_peers=thread_data.get("anonymous_to_peers", False), - closed=thread_data.get("closed", False), - closed_by=closed_by, - close_reason_code=thread_data.get("close_reason_code"), - pinned=thread_data.get("pinned", False), - created_at=parse_mongo_datetime(thread_data["created_at"]), - updated_at=parse_mongo_datetime(thread_data["updated_at"]), - last_activity_at=parse_mongo_datetime(thread_data["last_activity_at"]), - commentable_id=thread_data.get("commentable_id"), - # Moderation fields - is_spam=thread_data.get("is_spam", False), - is_deleted=thread_data.get("is_deleted", False), - deleted_at=parse_mongo_datetime(thread_data.get("deleted_at")), - deleted_by=deleted_by, - visible=thread_data.get("visible", True), - ) - mongo_content.content_object_id = thread.pk - mongo_content.content_type = thread.content_type - mongo_content.save() - else: - # Update existing thread with latest data from MongoDB - thread = CommentThread.objects.get(pk=mongo_content.content_object_id) - - deleted_by, closed_by = _resolve_thread_actors(thread_data) - - # Update all fields that might have changed - thread.title = get_trunc_title(thread_data.get("title", "")) - thread.body = thread_data["body"] - thread.thread_type = thread_data.get("thread_type", "discussion") - thread.context = thread_data.get("context", "course") - thread.anonymous = thread_data.get("anonymous", False) - thread.anonymous_to_peers = thread_data.get("anonymous_to_peers", False) - thread.closed = thread_data.get("closed", False) - thread.closed_by = closed_by # type: ignore[assignment] - thread.close_reason_code = thread_data.get("close_reason_code") - thread.pinned = thread_data.get("pinned", False) - thread.updated_at = parse_mongo_datetime(thread_data["updated_at"]) # type: ignore[assignment] - thread.last_activity_at = parse_mongo_datetime(thread_data["last_activity_at"]) - thread.commentable_id = thread_data.get("commentable_id") # type: ignore[assignment] - # Update moderation fields - thread.is_spam = thread_data.get("is_spam", False) - thread.is_deleted = thread_data.get("is_deleted", False) - thread.deleted_at = parse_mongo_datetime(thread_data.get("deleted_at")) - thread.deleted_by = deleted_by # type: ignore[assignment] - thread.visible = thread_data.get("visible", True) - thread.save() - - create_or_update_edit_history(thread_data) - create_or_update_abuse_flaggers(thread_data) - create_votes(thread, thread_data.get("votes", {})) - - -def create_or_update_comment( # pylint: disable=too-many-statements - comment_data: dict[str, Any], +def _bulk_migrate_comments( + comments_data: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + comment_ct: ContentType, ) -> None: - """Create or update a comment.""" - author = get_user_or_none(comment_data["author_id"]) - if not author: - return + """ + Bulk-create new comments and bulk-update existing ones. - # Preserve author_username from MongoDB (historical username) - author_username = comment_data.get("author_username") - - # Preserve retired_username from MongoDB - retired_username = comment_data.get("retired_username") - - # If author_username is not provided, use retired_username or fallback to current username - if not author_username: - if retired_username: - author_username = retired_username - else: - author_username = author.username - - mongo_thread_id = str(comment_data["comment_thread_id"]) - mongo_thread = MongoContent.objects.filter(mongo_id=mongo_thread_id).first() - if not mongo_thread: - logger.warning( - f"Thread mapping not found for comment {comment_data.get('_id')} " - f"(mongo_thread_id={mongo_thread_id})" - ) - return - thread = CommentThread.objects.filter(pk=mongo_thread.content_object_id).first() - if not thread: - logger.warning( - f"Skipping comment {comment_data.get('_id')}: thread object not found " - f"(content_object_id={mongo_thread.content_object_id})" - ) + Must be called with top-level comments before child comments so that + parent PKs are available in *mongo_cache* when children are processed. + """ + if not comments_data: return - parent = None - if "parent_id" in comment_data and comment_data["parent_id"] != "None": - parent_id = str(comment_data["parent_id"]) - mongo_parent_comment = MongoContent.objects.filter(mongo_id=parent_id).first() - if not mongo_parent_comment: - logger.warning( - f"Parent mapping not found for comment {comment_data.get('_id')} " - f"(parent_id={parent_id})" + + new_data = [ + c for c in comments_data + if not (mongo_cache.get(str(c["_id"])) and + mongo_cache[str(c["_id"])].content_object_id) + ] + existing_data = [ + c for c in comments_data + if mongo_cache.get(str(c["_id"])) and + mongo_cache[str(c["_id"])].content_object_id + ] + + # --- Create new comments in bulk --- + if new_data: + pairs: list[tuple[str, Comment]] = [] + for c in new_data: + author = user_cache.get(_to_int_id(c.get("author_id"))) # type: ignore[arg-type] + if not author: + continue + + mongo_thread_id = str(c["comment_thread_id"]) + mc_thread = mongo_cache.get(mongo_thread_id) + if not mc_thread or not mc_thread.content_object_id: + logger.warning( + f"Thread mapping not found for comment {c.get('_id')} " + f"(mongo_thread_id={mongo_thread_id})" + ) + continue + + parent_pk: int | None = None + if c.get("parent_id") and str(c.get("parent_id")) != "None": + mc_parent = mongo_cache.get(str(c["parent_id"])) + if not mc_parent or not mc_parent.content_object_id: + logger.warning( + f"Parent mapping not found for comment {c.get('_id')} " + f"(parent_id={c['parent_id']})" + ) + continue + parent_pk = mc_parent.content_object_id + + author_username = ( + c.get("author_username") + or c.get("retired_username") + or author.username ) - return - parent = Comment.objects.filter( - id=mongo_parent_comment.content_object_id - ).first() - if not parent: - logger.warning( - f"Skipping comment {comment_data.get('_id')}: parent object not found " - f"(parent_content_object_id={mongo_parent_comment.content_object_id})" + deleted_by = ( + user_cache.get(_to_int_id(c.get("deleted_by"))) # type: ignore[arg-type] + if c.get("deleted_by") else None ) - return + pairs.append(( + str(c["_id"]), + Comment( + author=author, + author_username=author_username, + retired_username=c.get("retired_username"), + comment_thread_id=mc_thread.content_object_id, + parent_id=parent_pk, + course_id=c["course_id"], + body=c["body"], + anonymous=c.get("anonymous", False), + anonymous_to_peers=c.get("anonymous_to_peers", False), + endorsed=c.get("endorsed", False), + child_count=c.get("child_count", 0), + created_at=parse_mongo_datetime(c["created_at"]), + updated_at=parse_mongo_datetime(c["updated_at"]), + depth=1 if parent_pk else 0, + is_spam=c.get("is_spam", False), + is_deleted=c.get("is_deleted", False), + deleted_at=parse_mongo_datetime(c.get("deleted_at")), + deleted_by=deleted_by, + visible=c.get("visible", True), + ), + )) + + if pairs: + mongo_ids, objs = zip(*pairs) + created = Comment.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) + + mc_rows = [] + sort_key_updates: list[Comment] = [] + for mid, comment in zip(mongo_ids, created): + if not comment.pk: + continue + mc_rows.append( + MongoContent( + mongo_id=mid, + content_type=comment_ct, + content_object_id=comment.pk, + ) + ) + # Set sort_key now that we have the PK. + if comment.parent_id: + comment.sort_key = f"{comment.parent_id}-{comment.pk}" + else: + comment.sort_key = f"{comment.pk}" + sort_key_updates.append(comment) + + if mc_rows: + MongoContent.objects.bulk_create( + mc_rows, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if sort_key_updates: + Comment.objects.bulk_update( + sort_key_updates, ["sort_key"], batch_size=BATCH_SIZE + ) - mongo_comment, _ = MongoContent.objects.get_or_create( - mongo_id=str(comment_data["_id"]) - ) - if not mongo_comment.content_object_id: - # Get deleted_by user if deleted_by field exists in MongoDB - deleted_by = None - if comment_data.get("deleted_by"): - deleted_by = get_user_or_none(comment_data["deleted_by"]) - - comment = Comment.objects.create( - author=author, - author_username=author_username, - retired_username=retired_username, - comment_thread=thread, - parent=parent, - course_id=comment_data["course_id"], - body=comment_data["body"], - anonymous=comment_data.get("anonymous", False), - anonymous_to_peers=comment_data.get("anonymous_to_peers", False), - endorsed=comment_data.get("endorsed", False), - child_count=comment_data.get("child_count", 0), - created_at=parse_mongo_datetime(comment_data["created_at"]), - updated_at=parse_mongo_datetime(comment_data["updated_at"]), - depth=1 if parent else 0, - # Moderation fields - is_spam=comment_data.get("is_spam", False), - is_deleted=comment_data.get("is_deleted", False), - deleted_at=parse_mongo_datetime(comment_data.get("deleted_at")), - deleted_by=deleted_by, - visible=comment_data.get("visible", True), - ) - mongo_comment.content_object_id = comment.pk - mongo_comment.content_type = comment.content_type - mongo_comment.save() - sort_key = f"{parent.pk}-{comment.pk}" if parent else f"{comment.pk}" - comment.sort_key = sort_key - comment.save() - else: - # Update existing comment with latest data from MongoDB - comment = Comment.objects.get(pk=mongo_comment.content_object_id) - - # Get deleted_by user if needed - deleted_by = None - if comment_data.get("deleted_by"): - deleted_by = get_user_or_none(comment_data["deleted_by"]) - - # Update all fields that might have changed - comment.body = comment_data["body"] - comment.anonymous = comment_data.get("anonymous", False) - comment.anonymous_to_peers = comment_data.get("anonymous_to_peers", False) - comment.endorsed = comment_data.get("endorsed", False) - comment.child_count = comment_data.get("child_count", 0) - comment.updated_at = parse_mongo_datetime(comment_data["updated_at"]) # type: ignore[assignment] - # Update moderation fields - comment.is_spam = comment_data.get("is_spam", False) - comment.is_deleted = comment_data.get("is_deleted", False) - comment.deleted_at = parse_mongo_datetime(comment_data.get("deleted_at")) - comment.deleted_by = deleted_by # type: ignore[assignment] - comment.visible = comment_data.get("visible", True) - comment.save() - - create_or_update_edit_history(comment_data) - create_or_update_abuse_flaggers(comment_data) - create_votes(comment, comment_data.get("votes", {})) - - -def create_votes(content: CommentThread | Comment, votes_data: dict[str, Any]) -> None: - """Create or update votes for a content.""" - for vote_type in ["up", "down"]: - for user_id in votes_data.get(vote_type, []): - user = get_user_or_none(user_id) - if not user: + # --- Update existing comments in bulk --- + if existing_data: + pks = [ + mongo_cache[str(c["_id"])].content_object_id + for c in existing_data + if str(c["_id"]) in mongo_cache + ] + comment_pk_map = { + cm.pk: cm for cm in Comment.objects.filter(pk__in=pks) + } + comment_update_fields = [ + "body", "anonymous", "anonymous_to_peers", "endorsed", "child_count", + "updated_at", "is_spam", "is_deleted", "deleted_at", "deleted_by", "visible", + ] + to_update: list[Comment] = [] + for c in existing_data: + mc = mongo_cache.get(str(c["_id"])) + if not mc: + continue + comment = comment_pk_map.get(mc.content_object_id) + if not comment: continue - UserVote.objects.update_or_create( - user=user, - content_type=content.content_type, - content_object_id=content.pk, - defaults={"vote": 1 if vote_type == "up" else -1}, + deleted_by = ( + user_cache.get(_to_int_id(c.get("deleted_by"))) # type: ignore[arg-type] + if c.get("deleted_by") else None + ) + comment.body = c["body"] + comment.anonymous = c.get("anonymous", False) + comment.anonymous_to_peers = c.get("anonymous_to_peers", False) + comment.endorsed = c.get("endorsed", False) + comment.child_count = c.get("child_count", 0) + comment.updated_at = parse_mongo_datetime(c["updated_at"]) # type: ignore[assignment] + comment.is_spam = c.get("is_spam", False) + comment.is_deleted = c.get("is_deleted", False) + comment.deleted_at = parse_mongo_datetime(c.get("deleted_at")) + comment.deleted_by = deleted_by # type: ignore[assignment] + comment.visible = c.get("visible", True) + to_update.append(comment) + if to_update: + Comment.objects.bulk_update( + to_update, comment_update_fields, batch_size=BATCH_SIZE ) -def create_or_update_edit_history(content: dict[str, Any]) -> None: - """Create or update edit history for a content.""" - edit_history = content.get("edit_history", []) - content_type = CommentThread if content["_type"] == "CommentThread" else Comment - mongo_content = MongoContent.objects.filter(mongo_id=str(content["_id"])).first() - if not mongo_content: - logger.warning( - f"Skipping edit history for content {content.get('_id')}: mapping not found" - ) +def _bulk_migrate_votes( + contents: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, + comment_ct: ContentType, +) -> None: + """Bulk-create missing UserVote rows for all content in one pass.""" + candidates: list[tuple[int, int, int, int]] = [] # (user_id, ct_id, obj_id, vote) + for c in contents: + mc = mongo_cache.get(str(c["_id"])) + if not mc or not mc.content_object_id: + continue + ct = thread_ct if c["_type"] == "CommentThread" else comment_ct + for vote_type in ("up", "down"): + vote_val = 1 if vote_type == "up" else -1 + for uid_str in c.get("votes", {}).get(vote_type, []): + uid = _to_int_id(uid_str) + if uid is not None and uid in user_cache: + candidates.append((uid, ct.pk, mc.content_object_id, vote_val)) + + if not candidates: return - content_object = content_type.objects.filter( - pk=mongo_content.content_object_id - ).first() - if not content_object: - logger.warning( - f"Skipping edit history for content {content.get('_id')}: target object not found " - f"(content_object_id={mongo_content.content_object_id})" + + obj_ids = {obj_id for _, _, obj_id, _ in candidates} + existing_keys = set( + UserVote.objects.filter(content_object_id__in=obj_ids).values_list( + "user_id", "content_type_id", "content_object_id" ) - return - for edit in edit_history: - editor = get_user_or_none(edit["author_id"]) - if not editor: + ) + + new_votes = [ + UserVote( + user_id=uid, + content_type_id=ct_id, + content_object_id=obj_id, + vote=vote_val, + ) + for uid, ct_id, obj_id, vote_val in candidates + if (uid, ct_id, obj_id) not in existing_keys + ] + if new_votes: + UserVote.objects.bulk_create( + new_votes, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + + +def _bulk_migrate_edit_history( + contents: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, + comment_ct: ContentType, +) -> None: + """Bulk-create missing EditHistory rows for all content in one pass.""" + obj_ids = [mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id] + existing_keys: set[tuple[int, int, Any, int]] = set( + EditHistory.objects.filter(content_object_id__in=obj_ids).values_list( + "content_object_id", "content_type_id", "created_at", "editor_id" + ) + ) + + to_create: list[EditHistory] = [] + for c in contents: + mc = mongo_cache.get(str(c["_id"])) + if not mc or not mc.content_object_id: continue - EditHistory.objects.get_or_create( - content_object_id=content_object.pk, - content_type=content_object.content_type, - created_at=parse_mongo_datetime(edit["created_at"]), - editor=editor, - defaults={ - "original_body": edit["original_body"], - "reason_code": edit["reason_code"], - }, + ct = thread_ct if c["_type"] == "CommentThread" else comment_ct + for edit in c.get("edit_history", []): + editor = user_cache.get(_to_int_id(edit.get("author_id"))) # type: ignore[arg-type] + if not editor: + continue + created_at = parse_mongo_datetime(edit["created_at"]) + key = (mc.content_object_id, ct.pk, created_at, editor.pk) + if key not in existing_keys: + to_create.append( + EditHistory( + content_object_id=mc.content_object_id, + content_type=ct, + created_at=created_at, + editor=editor, + original_body=edit["original_body"], + reason_code=edit["reason_code"], + ) + ) + existing_keys.add(key) # avoid duplicates within the same batch + + if to_create: + EditHistory.objects.bulk_create( + to_create, batch_size=BATCH_SIZE, ignore_conflicts=True ) -def create_or_update_abuse_flaggers(content: dict[str, Any]) -> None: - """Create or update abuse flaggers for content.""" - content_type = CommentThread if content["_type"] == "CommentThread" else Comment - mongo_content = MongoContent.objects.filter(mongo_id=str(content["_id"])).first() - if not mongo_content: - logger.warning( - f"Skipping abuse flaggers for content {content.get('_id')}: mapping not found" +def _bulk_migrate_abuse_flaggers( + contents: list[dict[str, Any]], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], + thread_ct: ContentType, + comment_ct: ContentType, +) -> None: + """Bulk-create missing AbuseFlagger / HistoricalAbuseFlagger rows.""" + obj_ids = [mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id] + + existing_af: set[tuple[int, int, int]] = set( + AbuseFlagger.objects.filter(content_object_id__in=obj_ids).values_list( + "user_id", "content_type_id", "content_object_id" ) - return - content_object = content_type.objects.filter( - pk=mongo_content.content_object_id - ).first() - if not content_object: - logger.warning( - f"Skipping abuse flaggers for content {content.get('_id')}: target object not found " - f"(content_object_id={mongo_content.content_object_id})" + ) + existing_haf: set[tuple[int, int, int]] = set( + HistoricalAbuseFlagger.objects.filter(content_object_id__in=obj_ids).values_list( + "user_id", "content_type_id", "content_object_id" ) - return - for user_id in content.get("abuse_flaggers", []): - user = get_user_or_none(user_id) - if not user: + ) + + af_to_create: list[AbuseFlagger] = [] + haf_to_create: list[HistoricalAbuseFlagger] = [] + flagged_at = timezone.now() + + for c in contents: + mc = mongo_cache.get(str(c["_id"])) + if not mc or not mc.content_object_id: continue - AbuseFlagger.objects.update_or_create( - user=user, - content_type=content_object.content_type, - content_object_id=content_object.pk, - defaults={ - "flagged_at": timezone.now(), - }, + ct = thread_ct if c["_type"] == "CommentThread" else comment_ct + + for uid_str in c.get("abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None and uid in user_cache: + key = (uid, ct.pk, mc.content_object_id) + if key not in existing_af: + af_to_create.append( + AbuseFlagger( + user_id=uid, + content_type=ct, + content_object_id=mc.content_object_id, + flagged_at=flagged_at, + ) + ) + existing_af.add(key) + + for uid_str in c.get("historical_abuse_flaggers", []): + uid = _to_int_id(uid_str) + if uid is not None and uid in user_cache: + key = (uid, ct.pk, mc.content_object_id) + if key not in existing_haf: + haf_to_create.append( + HistoricalAbuseFlagger( + user_id=uid, + content_type=ct, + content_object_id=mc.content_object_id, + flagged_at=flagged_at, + ) + ) + existing_haf.add(key) + + if af_to_create: + AbuseFlagger.objects.bulk_create( + af_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True ) - for user_id in content.get("historical_abuse_flaggers", []): - user = get_user_or_none(user_id) - if not user: - continue - HistoricalAbuseFlagger.objects.update_or_create( - user=user, - content_type=content_object.content_type, - content_object_id=content_object.pk, - defaults={ - "flagged_at": timezone.now(), - }, + if haf_to_create: + HistoricalAbuseFlagger.objects.bulk_create( + haf_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True ) -def migrate_subscriptions(db: Database[dict[str, Any]], content_id: str) -> None: - """Migrate subscriptions from mongo to mysql.""" - subscriptions = db.subscriptions.find({"source_id": str(content_id)}) - for sub in subscriptions: - user = get_user_or_none(sub["subscriber_id"]) +def _bulk_migrate_subscriptions( + db: Database[dict[str, Any]], + content_ids_str: list[str], + mongo_cache: dict[str, MongoContent], + user_cache: dict[int, User], +) -> None: + """ + Migrate subscriptions for an entire course using a SINGLE MongoDB query. + + The original code fired one ``db.subscriptions.find()`` per content + item. This version fetches them all at once via ``$in``. + """ + if not content_ids_str: + return + + all_subs = list( + db.subscriptions.find({"source_id": {"$in": content_ids_str}}) + ) + if not all_subs: + return + + thread_ct = ContentType.objects.get_for_model(CommentThread) + comment_ct = ContentType.objects.get_for_model(Comment) + + obj_ids = [mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id] + existing_sub_map: dict[tuple[int, int, int], Subscription] = { + (s.subscriber_id, s.source_content_type_id, s.source_object_id): s + for s in Subscription.objects.filter(source_object_id__in=obj_ids) + } + + subs_to_create: list[Subscription] = [] + subs_to_update: list[Subscription] = [] + seen: set[tuple[int, int, int]] = set(existing_sub_map.keys()) + + for sub in all_subs: + uid = _to_int_id(sub.get("subscriber_id")) + user = user_cache.get(uid) if uid is not None else None # type: ignore[arg-type] if not user: continue - content_type = ( - CommentThread if sub["source_type"] == "CommentThread" else Comment - ) - mongo_content = MongoContent.objects.filter(mongo_id=str(content_id)).first() - if not mongo_content: - logger.warning( - f"Skipping subscription for source {content_id}: mapping not found" - ) - continue - content = content_type.objects.filter( - pk=mongo_content.content_object_id - ).first() - if not content: + mc = mongo_cache.get(str(sub.get("source_id", ""))) + if not mc or not mc.content_object_id: logger.warning( - f"Skipping subscription for source {content_id}: target object not found " - f"(content_object_id={mongo_content.content_object_id})" + f"Skipping subscription for source {sub.get('source_id')}: mapping not found" ) continue - if content: - Subscription.objects.update_or_create( - subscriber=user, - source_content_type=content.content_type, - source_object_id=content.pk, - defaults={ - "created_at": parse_mongo_datetime(sub.get("created_at")) - or timezone.now(), - "updated_at": parse_mongo_datetime(sub.get("updated_at")) - or timezone.now(), - }, + ct = thread_ct if sub.get("source_type") == "CommentThread" else comment_ct + created_at = parse_mongo_datetime(sub.get("created_at")) or timezone.now() + updated_at = parse_mongo_datetime(sub.get("updated_at")) or timezone.now() + key = (user.pk, ct.pk, mc.content_object_id) + + if key in existing_sub_map: + s = existing_sub_map[key] + s.created_at = created_at # type: ignore[assignment] + s.updated_at = updated_at # type: ignore[assignment] + subs_to_update.append(s) + elif key not in seen: + subs_to_create.append( + Subscription( + subscriber=user, + source_content_type=ct, + source_object_id=mc.content_object_id, + created_at=created_at, + updated_at=updated_at, + ) ) + seen.add(key) + + if subs_to_create: + Subscription.objects.bulk_create( + subs_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if subs_to_update: + Subscription.objects.bulk_update( + subs_to_update, ["created_at", "updated_at"], batch_size=BATCH_SIZE + ) +# --------------------------------------------------------------------------- +# migrate_read_states (batch-optimised) +# --------------------------------------------------------------------------- + def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: - """Migrate read states from mongo to mysql.""" - users = db.users.find({"course_stats.course_id": course_id}) - for user_data in users: - user = get_user_or_none(user_data["_id"]) - if not user: + """ + Migrate read states from MongoDB to MySQL using bulk operations. + + Replaces the original per-row get_or_create / .save() loops with + one bulk-create pass for ReadState and one for LastReadTime. + """ + all_mongo_users = list(db.users.find({"course_stats.course_id": course_id})) + if not all_mongo_users: + return + + # Collect per-user read-state data and all referenced thread IDs. + all_thread_ids: set[str] = set() + user_read_data: list[tuple[int, list[dict[str, Any]]]] = [] + for user_data in all_mongo_users: + uid = _to_int_id(user_data["_id"]) + if uid is None: continue + relevant = [ + rs for rs in user_data.get("read_states", []) + if rs.get("course_id") == course_id + ] + if relevant: + user_read_data.append((uid, relevant)) + for rs in relevant: + all_thread_ids.update(rs.get("last_read_times", {}).keys()) + + if not user_read_data: + return + + uids = {uid for uid, _ in user_read_data} + django_users = {u.pk: u for u in User.objects.filter(pk__in=uids)} + + # Bulk-fetch all MongoContent for referenced thread IDs. + mongo_thread_map: dict[str, int] = {} # mongo_id → content_object_id + for mc in MongoContent.objects.filter(mongo_id__in=all_thread_ids): + if mc.content_object_id: + mongo_thread_map[mc.mongo_id] = mc.content_object_id + + # Bulk get-or-create ReadState rows. + existing_rs: dict[int, ReadState] = { + rs.user_id: rs + for rs in ReadState.objects.filter( + user_id__in=django_users.keys(), course_id=course_id + ) + } + new_rs = [ + ReadState(user_id=uid, course_id=course_id) + for uid, _ in user_read_data + if uid in django_users and uid not in existing_rs + ] + if new_rs: + ReadState.objects.bulk_create( + new_rs, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + existing_rs.update( + { + rs.user_id: rs + for rs in ReadState.objects.filter( + user_id__in=[r.user_id for r in new_rs], + course_id=course_id, + ) + } + ) + + # Bulk get-or-create LastReadTime rows. + rs_ids = [rs.pk for rs in existing_rs.values()] + existing_lrt: dict[tuple[int, int], LastReadTime] = { + (lrt.read_state_id, lrt.comment_thread_id): lrt + for lrt in LastReadTime.objects.filter(read_state_id__in=rs_ids) + } - for read_state in user_data.get("read_states", []): - if read_state["course_id"] == course_id: - rs, _ = ReadState.objects.get_or_create(user=user, course_id=course_id) - for thread_id, timestamp in read_state.get( - "last_read_times", {} - ).items(): - mongo_content = MongoContent.objects.filter( - mongo_id=thread_id - ).first() - thread = mongo_content and mongo_content.content - - # For older courses using cs_comment_service, the thread may be None - # because cs_comment_service retains read_states for deleted threads - # in the users collection in MongoDB. As a result, MongoContent won't - # have a thread with deleted thread_id from read_states. - if not thread: - continue - existing_read_time = LastReadTime.objects.filter( - read_state=rs, comment_thread=thread - ).first() - if not existing_read_time: - LastReadTime.objects.create( + lrt_to_create: list[LastReadTime] = [] + lrt_to_update: list[LastReadTime] = [] + seen_keys: set[tuple[int, int]] = set(existing_lrt.keys()) + + for uid, read_states in user_read_data: + if uid not in django_users: + continue + rs = existing_rs.get(uid) + if not rs: + continue + for read_state in read_states: + for thread_id, timestamp in read_state.get("last_read_times", {}).items(): + thread_pk = mongo_thread_map.get(thread_id) + # thread_pk may be None for deleted threads retained in MongoDB + # read_states — skip silently (same behaviour as original code). + if not thread_pk: + continue + parsed_ts = parse_mongo_datetime(timestamp) + key = (rs.pk, thread_pk) + if key in existing_lrt: + lrt = existing_lrt[key] + lrt.timestamp = parsed_ts # type: ignore[assignment] + lrt_to_update.append(lrt) + elif key not in seen_keys: + lrt_to_create.append( + LastReadTime( read_state=rs, - comment_thread=thread, - timestamp=parse_mongo_datetime(timestamp), + comment_thread_id=thread_pk, + timestamp=parsed_ts, ) - else: - existing_read_time.timestamp = parse_mongo_datetime(timestamp) # type: ignore[assignment] - existing_read_time.save() + ) + seen_keys.add(key) + + if lrt_to_create: + LastReadTime.objects.bulk_create( + lrt_to_create, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if lrt_to_update: + LastReadTime.objects.bulk_update( + lrt_to_update, ["timestamp"], batch_size=BATCH_SIZE + ) def delete_course_data( From 1c1b079360b94b25ae4ef19593223bad874d927b Mon Sep 17 00:00:00 2001 From: Alam-2U Date: Thu, 25 Jun 2026 07:16:19 +0000 Subject: [PATCH 2/4] fix: optimize forum migration to eliminate N+1 queries and improve scalability --- ...um_migrate_course_from_mongodb_to_mysql.py | 29 +- forum/migration_helpers.py | 394 ++++++++++++------ 2 files changed, 272 insertions(+), 151 deletions(-) diff --git a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py index 48b82fb0..ff779265 100644 --- a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py +++ b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py @@ -4,10 +4,10 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any -from django.core.management.base import BaseCommand -from django.core.management.base import CommandParser +from django.core.management.base import BaseCommand, CommandError, CommandParser from django.db import connections +import forum.migration_helpers as _migration_helpers from forum.migration_helpers import ( BATCH_SIZE, enable_mysql_backend_for_course, @@ -37,10 +37,6 @@ def _migrate_one_course( Returns ``(course_id, elapsed_seconds, error_message_or_None)``. Each thread gets its own MongoDB client and Django DB connection. """ - # Ensure this thread does not reuse a connection inherited from the - # main thread (Django creates a new one on first access per thread). - connections.close_all() - # Django stores DB connections in thread-local storage, so each worker # thread automatically gets its own connection on first access. # Explicitly close any connection inherited from the spawning thread @@ -108,11 +104,16 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: db = get_database() create_waffle_flags = not options["no_toggle"] - workers: int = int(options["workers"]) # type: ignore[arg-type] + workers: int = int(str(options["workers"])) + batch_size: int = int(str(options["batch_size"])) + + if workers < 1: + raise CommandError("--workers must be >= 1.") + if batch_size < 1: + raise CommandError("--batch-size must be >= 1.") # Override module-level BATCH_SIZE when caller passes --batch-size. - import forum.migration_helpers as _mh - _mh.BATCH_SIZE = int(options["batch_size"]) # type: ignore[arg-type] + _migration_helpers.BATCH_SIZE = batch_size course_ids = list(options["courses"]) if "all" in course_ids: @@ -121,7 +122,7 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: total = len(course_ids) self.stdout.write( f"Migrating {total} course(s) with {workers} parallel worker(s) " - f"(batch_size={_mh.BATCH_SIZE})." + f"(batch_size={_migration_helpers.BATCH_SIZE})." ) failed: list[tuple[str, str]] = [] @@ -167,12 +168,12 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: if failed: self.stderr.write( - self.style.ERROR( - f"\n{len(failed)} course(s) failed migration:" - ) + self.style.ERROR(f"\n{len(failed)} course(s) failed migration:") ) for cid, err in failed: self.stderr.write(self.style.ERROR(f" {cid}: {err}")) - raise SystemExit(1) + raise CommandError( + f"{len(failed)} course(s) failed migration. See stderr for details." + ) self.stdout.write(self.style.SUCCESS("Data migration completed successfully")) diff --git a/forum/migration_helpers.py b/forum/migration_helpers.py index eb64cae2..b87aec0b 100644 --- a/forum/migration_helpers.py +++ b/forum/migration_helpers.py @@ -2,11 +2,12 @@ import logging from datetime import datetime -from typing import Any +from typing import Any, cast from django.contrib.auth.models import User # pylint: disable=E5142 from django.contrib.contenttypes.models import ContentType from django.core.management.base import OutputWrapper +from django.db.models import Max from django.utils import timezone from pymongo.collection import Collection from pymongo.database import Database @@ -70,6 +71,7 @@ def get_all_course_ids(db: Database[dict[str, Any]]) -> list[str]: # Internal helpers # --------------------------------------------------------------------------- + def _to_int_id(uid: Any) -> int | None: """Convert a user ID value to int; return None on failure.""" try: @@ -117,6 +119,7 @@ def _build_user_cache(user_ids: set[int]) -> dict[int, User]: # migrate_users (batch-optimised) # --------------------------------------------------------------------------- + def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: """ Migrate users from MongoDB to MySQL. @@ -162,14 +165,21 @@ def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: # --- CourseStat: bulk create new / bulk update existing --- existing_stats: dict[int, CourseStat] = { - cs.user_id: cs + cast(int, cs.user_id): cs # type: ignore[attr-defined] for cs in CourseStat.objects.filter( user_id__in=django_users.keys(), course_id=course_id ) } stat_update_fields = [ - "active_flags", "inactive_flags", "threads", "responses", "replies", - "deleted_threads", "deleted_responses", "deleted_replies", "last_activity_at", + "active_flags", + "inactive_flags", + "threads", + "responses", + "replies", + "deleted_threads", + "deleted_responses", + "deleted_replies", + "last_activity_at", ] stats_to_create: list[CourseStat] = [] stats_to_update: list[CourseStat] = [] @@ -224,6 +234,7 @@ def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: # migrate_content (batch-optimised) # --------------------------------------------------------------------------- + def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: """ Migrate threads and comments from MongoDB to MySQL. @@ -239,9 +250,7 @@ def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: 6. Issue a SINGLE MongoDB ``$in`` query for all subscriptions instead of one query per content item. """ - contents = list( - db.contents.find({"course_id": course_id}).sort("created_at", 1) - ) + contents = list(db.contents.find({"course_id": course_id}).sort("created_at", 1)) if not contents: return @@ -249,8 +258,7 @@ def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: # Pre-fetch all existing MongoContent rows (one query). mongo_cache: dict[str, MongoContent] = { - mc.mongo_id: mc - for mc in MongoContent.objects.filter(mongo_id__in=all_ids_str) + mc.mongo_id: mc for mc in MongoContent.objects.filter(mongo_id__in=all_ids_str) } # Collect every user ID referenced anywhere; fetch them all at once. @@ -269,11 +277,13 @@ def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: # --- Comments: top-level first, then children (parent must exist first) --- top_level = [ - c for c in comments_data + c + for c in comments_data if not c.get("parent_id") or str(c.get("parent_id")) == "None" ] child_comments = [ - c for c in comments_data + c + for c in comments_data if c.get("parent_id") and str(c.get("parent_id")) != "None" ] _bulk_migrate_comments(top_level, mongo_cache, user_cache, comment_ct) @@ -284,15 +294,15 @@ def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: # --- Metadata: votes, edit history, flaggers --- _bulk_migrate_votes(contents, mongo_cache, user_cache, thread_ct, comment_ct) _bulk_migrate_edit_history(contents, mongo_cache, user_cache, thread_ct, comment_ct) - _bulk_migrate_abuse_flaggers(contents, mongo_cache, user_cache, thread_ct, comment_ct) + _bulk_migrate_abuse_flaggers( + contents, mongo_cache, user_cache, thread_ct, comment_ct + ) # --- Subscriptions: ONE MongoDB query for the entire course --- _bulk_migrate_subscriptions(db, all_ids_str, mongo_cache, user_cache) -def _refresh_mongo_cache( - cache: dict[str, MongoContent], mongo_ids: list[str] -) -> None: +def _refresh_mongo_cache(cache: dict[str, MongoContent], mongo_ids: list[str]) -> None: """Update *cache* with freshly queried MongoContent rows for *mongo_ids*.""" if not mongo_ids: return @@ -301,7 +311,7 @@ def _refresh_mongo_cache( ) -def _bulk_migrate_threads( +def _bulk_migrate_threads( # pylint: disable=too-many-statements threads_data: list[dict[str, Any]], mongo_cache: dict[str, MongoContent], user_cache: dict[int, User], @@ -312,14 +322,18 @@ def _bulk_migrate_threads( return new_data = [ - t for t in threads_data - if not (mongo_cache.get(str(t["_id"])) and - mongo_cache[str(t["_id"])].content_object_id) + t + for t in threads_data + if not ( + mongo_cache.get(str(t["_id"])) + and mongo_cache[str(t["_id"])].content_object_id + ) ] existing_data = [ - t for t in threads_data - if mongo_cache.get(str(t["_id"])) and - mongo_cache[str(t["_id"])].content_object_id + t + for t in threads_data + if mongo_cache.get(str(t["_id"])) + and mongo_cache[str(t["_id"])].content_object_id ] # --- Create new threads in bulk --- @@ -330,51 +344,61 @@ def _bulk_migrate_threads( if not author: continue author_username = ( - t.get("author_username") - or t.get("retired_username") - or author.username + t.get("author_username") or t.get("retired_username") or author.username ) deleted_by = ( user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] - if t.get("deleted_by") else None + if t.get("deleted_by") + else None ) closed_by = ( user_cache.get(_to_int_id(t.get("closed_by_id"))) # type: ignore[arg-type] - if t.get("closed_by_id") else None + if t.get("closed_by_id") + else None + ) + pairs.append( + ( + str(t["_id"]), + CommentThread( + author=author, + author_username=author_username, + retired_username=t.get("retired_username"), + course_id=t["course_id"], + title=get_trunc_title(t.get("title", "")), + body=t["body"], + thread_type=t.get("thread_type", "discussion"), + context=t.get("context", "course"), + anonymous=t.get("anonymous", False), + anonymous_to_peers=t.get("anonymous_to_peers", False), + closed=t.get("closed", False), + closed_by=closed_by, + close_reason_code=t.get("close_reason_code"), + pinned=t.get("pinned", False), + created_at=parse_mongo_datetime(t["created_at"]), + updated_at=parse_mongo_datetime(t["updated_at"]), + last_activity_at=parse_mongo_datetime(t["last_activity_at"]), + commentable_id=t.get("commentable_id"), + is_spam=t.get("is_spam", False), + is_deleted=t.get("is_deleted", False), + deleted_at=parse_mongo_datetime(t.get("deleted_at")), + deleted_by=deleted_by, + visible=t.get("visible", True), + ), + ) ) - pairs.append(( - str(t["_id"]), - CommentThread( - author=author, - author_username=author_username, - retired_username=t.get("retired_username"), - course_id=t["course_id"], - title=get_trunc_title(t.get("title", "")), - body=t["body"], - thread_type=t.get("thread_type", "discussion"), - context=t.get("context", "course"), - anonymous=t.get("anonymous", False), - anonymous_to_peers=t.get("anonymous_to_peers", False), - closed=t.get("closed", False), - closed_by=closed_by, - close_reason_code=t.get("close_reason_code"), - pinned=t.get("pinned", False), - created_at=parse_mongo_datetime(t["created_at"]), - updated_at=parse_mongo_datetime(t["updated_at"]), - last_activity_at=parse_mongo_datetime(t["last_activity_at"]), - commentable_id=t.get("commentable_id"), - is_spam=t.get("is_spam", False), - is_deleted=t.get("is_deleted", False), - deleted_at=parse_mongo_datetime(t.get("deleted_at")), - deleted_by=deleted_by, - visible=t.get("visible", True), - ), - )) if pairs: mongo_ids, objs = zip(*pairs) - # Django 4.1+ returns PKs from bulk_create on MySQL 8.0.19+. - created = CommentThread.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) + + # MySQL's bulk_create does not always return PKs (Django feature + # flag can_return_rows_from_bulk_insert may be False). Snapshot + # the current max PK so we can re-fetch created rows afterwards. + # This is safe because migration runs as a single writer. + max_pk_before = CommentThread.objects.aggregate(Max("pk"))["pk__max"] or 0 + CommentThread.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) + created = list( + CommentThread.objects.filter(pk__gt=max_pk_before).order_by("pk") + ) mc_rows = [ MongoContent( mongo_id=mid, @@ -385,9 +409,37 @@ def _bulk_migrate_threads( if thread.pk ] if mc_rows: - MongoContent.objects.bulk_create( - mc_rows, batch_size=BATCH_SIZE, ignore_conflicts=True - ) + # Idempotent upsert for MongoContent mappings: + # - Rows that already exist with content_object_id=NULL (partial + # prior run) are updated via bulk_update. + # - Genuinely new rows are inserted via bulk_create. + # This avoids update_conflicts/unique_fields which is not + # supported on MySQL. + mc_row_map = {r.mongo_id: r for r in mc_rows} + existing_null = { + mc.mongo_id: mc + for mc in MongoContent.objects.filter( + mongo_id__in=list(mc_row_map), + content_object_id__isnull=True, + ) + } + to_update_mc = [] + for mongo_id, existing in existing_null.items(): + new = mc_row_map[mongo_id] + existing.content_type = new.content_type # type: ignore[assignment] + existing.content_object_id = new.content_object_id + to_update_mc.append(existing) + truly_new = [r for r in mc_rows if r.mongo_id not in existing_null] + if truly_new: + MongoContent.objects.bulk_create( + truly_new, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if to_update_mc: + MongoContent.objects.bulk_update( + to_update_mc, + ["content_type", "content_object_id"], + batch_size=BATCH_SIZE, + ) # --- Update existing threads in bulk --- if existing_data: @@ -396,15 +448,26 @@ def _bulk_migrate_threads( for t in existing_data if str(t["_id"]) in mongo_cache ] - thread_pk_map = { - th.pk: th - for th in CommentThread.objects.filter(pk__in=pks) - } + thread_pk_map = {th.pk: th for th in CommentThread.objects.filter(pk__in=pks)} thread_update_fields = [ - "title", "body", "thread_type", "context", "anonymous", - "anonymous_to_peers", "closed", "closed_by", "close_reason_code", - "pinned", "updated_at", "last_activity_at", "commentable_id", - "is_spam", "is_deleted", "deleted_at", "deleted_by", "visible", + "title", + "body", + "thread_type", + "context", + "anonymous", + "anonymous_to_peers", + "closed", + "closed_by", + "close_reason_code", + "pinned", + "updated_at", + "last_activity_at", + "commentable_id", + "is_spam", + "is_deleted", + "deleted_at", + "deleted_by", + "visible", ] to_update: list[CommentThread] = [] for t in existing_data: @@ -416,11 +479,13 @@ def _bulk_migrate_threads( continue deleted_by = ( user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] - if t.get("deleted_by") else None + if t.get("deleted_by") + else None ) closed_by = ( user_cache.get(_to_int_id(t.get("closed_by_id"))) # type: ignore[arg-type] - if t.get("closed_by_id") else None + if t.get("closed_by_id") + else None ) thread.title = get_trunc_title(t.get("title", "")) thread.body = t["body"] @@ -447,7 +512,7 @@ def _bulk_migrate_threads( ) -def _bulk_migrate_comments( +def _bulk_migrate_comments( # pylint: disable=too-many-statements comments_data: list[dict[str, Any]], mongo_cache: dict[str, MongoContent], user_cache: dict[int, User], @@ -463,14 +528,18 @@ def _bulk_migrate_comments( return new_data = [ - c for c in comments_data - if not (mongo_cache.get(str(c["_id"])) and - mongo_cache[str(c["_id"])].content_object_id) + c + for c in comments_data + if not ( + mongo_cache.get(str(c["_id"])) + and mongo_cache[str(c["_id"])].content_object_id + ) ] existing_data = [ - c for c in comments_data - if mongo_cache.get(str(c["_id"])) and - mongo_cache[str(c["_id"])].content_object_id + c + for c in comments_data + if mongo_cache.get(str(c["_id"])) + and mongo_cache[str(c["_id"])].content_object_id ] # --- Create new comments in bulk --- @@ -502,42 +571,48 @@ def _bulk_migrate_comments( parent_pk = mc_parent.content_object_id author_username = ( - c.get("author_username") - or c.get("retired_username") - or author.username + c.get("author_username") or c.get("retired_username") or author.username ) deleted_by = ( user_cache.get(_to_int_id(c.get("deleted_by"))) # type: ignore[arg-type] - if c.get("deleted_by") else None + if c.get("deleted_by") + else None + ) + pairs.append( + ( + str(c["_id"]), + Comment( + author=author, + author_username=author_username, + retired_username=c.get("retired_username"), + comment_thread_id=mc_thread.content_object_id, + parent_id=parent_pk, + course_id=c["course_id"], + body=c["body"], + anonymous=c.get("anonymous", False), + anonymous_to_peers=c.get("anonymous_to_peers", False), + endorsed=c.get("endorsed", False), + child_count=c.get("child_count", 0), + created_at=parse_mongo_datetime(c["created_at"]), + updated_at=parse_mongo_datetime(c["updated_at"]), + depth=1 if parent_pk else 0, + is_spam=c.get("is_spam", False), + is_deleted=c.get("is_deleted", False), + deleted_at=parse_mongo_datetime(c.get("deleted_at")), + deleted_by=deleted_by, + visible=c.get("visible", True), + ), + ) ) - pairs.append(( - str(c["_id"]), - Comment( - author=author, - author_username=author_username, - retired_username=c.get("retired_username"), - comment_thread_id=mc_thread.content_object_id, - parent_id=parent_pk, - course_id=c["course_id"], - body=c["body"], - anonymous=c.get("anonymous", False), - anonymous_to_peers=c.get("anonymous_to_peers", False), - endorsed=c.get("endorsed", False), - child_count=c.get("child_count", 0), - created_at=parse_mongo_datetime(c["created_at"]), - updated_at=parse_mongo_datetime(c["updated_at"]), - depth=1 if parent_pk else 0, - is_spam=c.get("is_spam", False), - is_deleted=c.get("is_deleted", False), - deleted_at=parse_mongo_datetime(c.get("deleted_at")), - deleted_by=deleted_by, - visible=c.get("visible", True), - ), - )) if pairs: mongo_ids, objs = zip(*pairs) - created = Comment.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) + + # Same max-PK-before trick as for threads: MySQL may not return + # PKs from bulk_create, so we re-fetch created rows by pk range. + max_pk_before = Comment.objects.aggregate(Max("pk"))["pk__max"] or 0 + Comment.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) + created = list(Comment.objects.filter(pk__gt=max_pk_before).order_by("pk")) mc_rows = [] sort_key_updates: list[Comment] = [] @@ -552,16 +627,40 @@ def _bulk_migrate_comments( ) ) # Set sort_key now that we have the PK. - if comment.parent_id: - comment.sort_key = f"{comment.parent_id}-{comment.pk}" + parent_id = cast(int | None, comment.parent_id) # type: ignore[attr-defined] + if parent_id: + comment.sort_key = f"{parent_id}-{comment.pk}" else: comment.sort_key = f"{comment.pk}" sort_key_updates.append(comment) if mc_rows: - MongoContent.objects.bulk_create( - mc_rows, batch_size=BATCH_SIZE, ignore_conflicts=True - ) + # Same idempotent upsert as for threads. + mc_row_map = {r.mongo_id: r for r in mc_rows} + existing_null = { + mc.mongo_id: mc + for mc in MongoContent.objects.filter( + mongo_id__in=list(mc_row_map), + content_object_id__isnull=True, + ) + } + to_update_mc = [] + for mongo_id, existing in existing_null.items(): + new = mc_row_map[mongo_id] + existing.content_type = new.content_type # type: ignore[assignment] + existing.content_object_id = new.content_object_id + to_update_mc.append(existing) + truly_new = [r for r in mc_rows if r.mongo_id not in existing_null] + if truly_new: + MongoContent.objects.bulk_create( + truly_new, batch_size=BATCH_SIZE, ignore_conflicts=True + ) + if to_update_mc: + MongoContent.objects.bulk_update( + to_update_mc, + ["content_type", "content_object_id"], + batch_size=BATCH_SIZE, + ) if sort_key_updates: Comment.objects.bulk_update( sort_key_updates, ["sort_key"], batch_size=BATCH_SIZE @@ -574,24 +673,32 @@ def _bulk_migrate_comments( for c in existing_data if str(c["_id"]) in mongo_cache ] - comment_pk_map = { - cm.pk: cm for cm in Comment.objects.filter(pk__in=pks) - } + comment_pk_map = {cm.pk: cm for cm in Comment.objects.filter(pk__in=pks)} comment_update_fields = [ - "body", "anonymous", "anonymous_to_peers", "endorsed", "child_count", - "updated_at", "is_spam", "is_deleted", "deleted_at", "deleted_by", "visible", + "body", + "anonymous", + "anonymous_to_peers", + "endorsed", + "child_count", + "updated_at", + "is_spam", + "is_deleted", + "deleted_at", + "deleted_by", + "visible", ] to_update: list[Comment] = [] for c in existing_data: mc = mongo_cache.get(str(c["_id"])) if not mc: continue - comment = comment_pk_map.get(mc.content_object_id) + comment: Comment | None = comment_pk_map.get(mc.content_object_id) # type: ignore[no-redef] if not comment: continue deleted_by = ( user_cache.get(_to_int_id(c.get("deleted_by"))) # type: ignore[arg-type] - if c.get("deleted_by") else None + if c.get("deleted_by") + else None ) comment.body = c["body"] comment.anonymous = c.get("anonymous", False) @@ -666,7 +773,9 @@ def _bulk_migrate_edit_history( comment_ct: ContentType, ) -> None: """Bulk-create missing EditHistory rows for all content in one pass.""" - obj_ids = [mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id] + obj_ids = [ + mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id + ] existing_keys: set[tuple[int, int, Any, int]] = set( EditHistory.objects.filter(content_object_id__in=obj_ids).values_list( "content_object_id", "content_type_id", "created_at", "editor_id" @@ -712,7 +821,9 @@ def _bulk_migrate_abuse_flaggers( comment_ct: ContentType, ) -> None: """Bulk-create missing AbuseFlagger / HistoricalAbuseFlagger rows.""" - obj_ids = [mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id] + obj_ids = [ + mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id + ] existing_af: set[tuple[int, int, int]] = set( AbuseFlagger.objects.filter(content_object_id__in=obj_ids).values_list( @@ -720,9 +831,9 @@ def _bulk_migrate_abuse_flaggers( ) ) existing_haf: set[tuple[int, int, int]] = set( - HistoricalAbuseFlagger.objects.filter(content_object_id__in=obj_ids).values_list( - "user_id", "content_type_id", "content_object_id" - ) + HistoricalAbuseFlagger.objects.filter( + content_object_id__in=obj_ids + ).values_list("user_id", "content_type_id", "content_object_id") ) af_to_create: list[AbuseFlagger] = [] @@ -790,18 +901,22 @@ def _bulk_migrate_subscriptions( if not content_ids_str: return - all_subs = list( - db.subscriptions.find({"source_id": {"$in": content_ids_str}}) - ) + all_subs = list(db.subscriptions.find({"source_id": {"$in": content_ids_str}})) if not all_subs: return thread_ct = ContentType.objects.get_for_model(CommentThread) comment_ct = ContentType.objects.get_for_model(Comment) - obj_ids = [mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id] + obj_ids = [ + mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id + ] existing_sub_map: dict[tuple[int, int, int], Subscription] = { - (s.subscriber_id, s.source_content_type_id, s.source_object_id): s + ( + cast(int, s.subscriber_id), # type: ignore[attr-defined] + cast(int, s.source_content_type_id), # type: ignore[attr-defined] + s.source_object_id, + ): s for s in Subscription.objects.filter(source_object_id__in=obj_ids) } @@ -811,7 +926,7 @@ def _bulk_migrate_subscriptions( for sub in all_subs: uid = _to_int_id(sub.get("subscriber_id")) - user = user_cache.get(uid) if uid is not None else None # type: ignore[arg-type] + user = user_cache.get(uid) if uid is not None else None if not user: continue mc = mongo_cache.get(str(sub.get("source_id", ""))) @@ -827,8 +942,8 @@ def _bulk_migrate_subscriptions( if key in existing_sub_map: s = existing_sub_map[key] - s.created_at = created_at # type: ignore[assignment] - s.updated_at = updated_at # type: ignore[assignment] + s.created_at = created_at + s.updated_at = updated_at subs_to_update.append(s) elif key not in seen: subs_to_create.append( @@ -856,6 +971,7 @@ def _bulk_migrate_subscriptions( # migrate_read_states (batch-optimised) # --------------------------------------------------------------------------- + def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: """ Migrate read states from MongoDB to MySQL using bulk operations. @@ -875,7 +991,8 @@ def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: if uid is None: continue relevant = [ - rs for rs in user_data.get("read_states", []) + rs + for rs in user_data.get("read_states", []) if rs.get("course_id") == course_id ] if relevant: @@ -897,13 +1014,13 @@ def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: # Bulk get-or-create ReadState rows. existing_rs: dict[int, ReadState] = { - rs.user_id: rs + cast(int, rs.user_id): rs # type: ignore[attr-defined] for rs in ReadState.objects.filter( user_id__in=django_users.keys(), course_id=course_id ) } new_rs = [ - ReadState(user_id=uid, course_id=course_id) + ReadState(user=django_users[uid], course_id=course_id) for uid, _ in user_read_data if uid in django_users and uid not in existing_rs ] @@ -913,9 +1030,9 @@ def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: ) existing_rs.update( { - rs.user_id: rs + cast(int, rs.user_id): rs # type: ignore[attr-defined] for rs in ReadState.objects.filter( - user_id__in=[r.user_id for r in new_rs], + user__in=[r.user for r in new_rs], course_id=course_id, ) } @@ -924,7 +1041,10 @@ def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: # Bulk get-or-create LastReadTime rows. rs_ids = [rs.pk for rs in existing_rs.values()] existing_lrt: dict[tuple[int, int], LastReadTime] = { - (lrt.read_state_id, lrt.comment_thread_id): lrt + ( + cast(int, lrt.read_state_id), # type: ignore[attr-defined] + cast(int, lrt.comment_thread_id), # type: ignore[attr-defined] + ): lrt for lrt in LastReadTime.objects.filter(read_state_id__in=rs_ids) } From 06f0f32aade1fe5832311639c050299045b8c4bb Mon Sep 17 00:00:00 2001 From: Alam-2U Date: Fri, 17 Jul 2026 11:40:12 +0000 Subject: [PATCH 3/4] fix: optimize forum migration script --- ...um_migrate_course_from_mongodb_to_mysql.py | 53 ++- forum/migration_helpers.py | 439 ++++++++++++------ .../test_commands/test_migration_commands.py | 289 ++++++++++++ 3 files changed, 635 insertions(+), 146 deletions(-) diff --git a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py index ff779265..25c86fdb 100644 --- a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py +++ b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py @@ -1,11 +1,13 @@ """Migration command for courses from mongodb to mysql.""" +from datetime import datetime import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any from django.core.management.base import BaseCommand, CommandError, CommandParser from django.db import connections +from django.utils import timezone import forum.migration_helpers as _migration_helpers from forum.migration_helpers import ( @@ -30,6 +32,7 @@ def _migrate_one_course( course_id: str, create_waffle_flags: bool, + updated_since: datetime | None = None, ) -> tuple[str, float | None, str | None]: """ Migrate a single course in the calling thread. @@ -46,8 +49,8 @@ def _migrate_one_course( db = get_database() t0 = time.monotonic() try: - migrate_users(db, course_id) - migrate_content(db, course_id) + migrate_users(db, course_id, updated_since=updated_since) + migrate_content(db, course_id, updated_since=updated_since) migrate_read_states(db, course_id) if create_waffle_flags: enable_mysql_backend_for_course(course_id) @@ -66,6 +69,23 @@ class Command(BaseCommand): help = "Migrate data from MongoDB to MySQL" + @staticmethod + def _parse_updated_since(raw: str | None) -> datetime | None: + """Parse an ISO8601 timestamp for incremental migration filtering.""" + if not raw: + return None + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError as exc: + raise CommandError( + "--updated-since must be a valid ISO8601 datetime, " + "for example: 2026-07-17T00:00:00Z" + ) from exc + + if timezone.is_naive(parsed): + parsed = timezone.make_aware(parsed, timezone=timezone.utc) + return parsed + def add_arguments(self, parser: CommandParser) -> None: """Add arguments to the command.""" parser.add_argument( @@ -95,6 +115,16 @@ def add_arguments(self, parser: CommandParser) -> None: metavar="N", help=f"Bulk-operation batch size (default: {BATCH_SIZE}).", ) + parser.add_argument( + "--updated-since", + type=str, + default=None, + help=( + "Optional ISO8601 datetime filter. Only Mongo records updated " + "on/after this time are migrated for users/content/subscriptions. " + "Example: 2026-07-17T00:00:00Z" + ), + ) parser.add_argument( "courses", nargs="+", type=str, help="List of course IDs or `all`" ) @@ -106,6 +136,7 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: create_waffle_flags = not options["no_toggle"] workers: int = int(str(options["workers"])) batch_size: int = int(str(options["batch_size"])) + updated_since = self._parse_updated_since(options.get("updated_since")) if workers < 1: raise CommandError("--workers must be >= 1.") @@ -120,10 +151,13 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: course_ids = get_all_course_ids(db) total = len(course_ids) - self.stdout.write( + run_msg = ( f"Migrating {total} course(s) with {workers} parallel worker(s) " f"(batch_size={_migration_helpers.BATCH_SIZE})." ) + if updated_since is not None: + run_msg += f" updated_since={updated_since.isoformat()}" + self.stdout.write(run_msg) failed: list[tuple[str, str]] = [] completed = 0 @@ -131,7 +165,11 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: if workers == 1: # Single-threaded path: simpler, no executor overhead. for course_id in course_ids: - cid, elapsed, err = _migrate_one_course(course_id, create_waffle_flags) + cid, elapsed, err = _migrate_one_course( + course_id, + create_waffle_flags, + updated_since=updated_since, + ) completed += 1 if err: self.stderr.write( @@ -147,7 +185,12 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: else: with ThreadPoolExecutor(max_workers=workers) as pool: futures = { - pool.submit(_migrate_one_course, cid, create_waffle_flags): cid + pool.submit( + _migrate_one_course, + cid, + create_waffle_flags, + updated_since, + ): cid for cid in course_ids } for future in as_completed(futures): diff --git a/forum/migration_helpers.py b/forum/migration_helpers.py index b87aec0b..55e21dec 100644 --- a/forum/migration_helpers.py +++ b/forum/migration_helpers.py @@ -115,19 +115,44 @@ def _build_user_cache(user_ids: set[int]) -> dict[int, User]: return {u.pk: u for u in User.objects.filter(pk__in=user_ids)} +def _set_if_changed(obj: Any, field: str, value: Any) -> bool: + """Set obj. only when value differs; return whether a change happened.""" + if getattr(obj, field) != value: + setattr(obj, field, value) + return True + return False + + # --------------------------------------------------------------------------- # migrate_users (batch-optimised) # --------------------------------------------------------------------------- -def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: +def migrate_users( + db: Database[dict[str, Any]], + course_id: str, + updated_since: datetime | None = None, +) -> None: """ Migrate users from MongoDB to MySQL. Uses bulk_create / bulk_update instead of per-row get_or_create calls, reducing the number of SQL round-trips from O(N) to O(1). """ - all_mongo_users = list(db.users.find({"course_stats.course_id": course_id})) + users_query: dict[str, Any] + if updated_since is not None: + users_query = { + "course_stats": { + "$elemMatch": { + "course_id": course_id, + "last_activity_at": {"$gte": updated_since}, + } + } + } + else: + users_query = {"course_stats.course_id": course_id} + + all_mongo_users = list(db.users.find(users_query)) if not all_mongo_users: return @@ -144,24 +169,27 @@ def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: # --- Single bulk fetch of all Django users for this course --- django_users = {u.pk: u for u in User.objects.filter(pk__in=uid_map.keys())} - # --- ForumUser: create missing rows in one shot --- - existing_fu_ids = set( - ForumUser.objects.filter(user_id__in=django_users.keys()).values_list( - "user_id", flat=True - ) - ) - new_forum_users = [ - ForumUser( - user_id=uid, - default_sort_key=uid_map[uid].get("default_sort_key", "date"), - ) - for uid in django_users - if uid not in existing_fu_ids - ] + # --- ForumUser: create missing rows; update default_sort_key if changed --- + existing_fu_map: dict[int, ForumUser] = { + fu.user_id: fu # type: ignore[attr-defined] + for fu in ForumUser.objects.filter(user_id__in=django_users.keys()) + } + new_forum_users = [] + fu_to_update: list[ForumUser] = [] + for uid in django_users: + sort_key = uid_map[uid].get("default_sort_key", "date") + if uid not in existing_fu_map: + new_forum_users.append(ForumUser(user_id=uid, default_sort_key=sort_key)) + else: + fu = existing_fu_map[uid] + if _set_if_changed(fu, "default_sort_key", sort_key): + fu_to_update.append(fu) if new_forum_users: ForumUser.objects.bulk_create( new_forum_users, batch_size=BATCH_SIZE, ignore_conflicts=True ) + if fu_to_update: + ForumUser.objects.bulk_update(fu_to_update, ["default_sort_key"], batch_size=BATCH_SIZE) # --- CourseStat: bulk create new / bulk update existing --- existing_stats: dict[int, CourseStat] = { @@ -193,16 +221,18 @@ def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: last_activity_at = parse_mongo_datetime(stat.get("last_activity_at")) if uid in existing_stats: cs = existing_stats[uid] - cs.active_flags = stat.get("active_flags", 0) - cs.inactive_flags = stat.get("inactive_flags", 0) - cs.threads = stat.get("threads", 0) - cs.responses = stat.get("responses", 0) - cs.replies = stat.get("replies", 0) - cs.deleted_threads = stat.get("deleted_threads", 0) - cs.deleted_responses = stat.get("deleted_responses", 0) - cs.deleted_replies = stat.get("deleted_replies", 0) - cs.last_activity_at = last_activity_at - stats_to_update.append(cs) + stat_changed = False + stat_changed |= _set_if_changed(cs, "active_flags", stat.get("active_flags", 0)) + stat_changed |= _set_if_changed(cs, "inactive_flags", stat.get("inactive_flags", 0)) + stat_changed |= _set_if_changed(cs, "threads", stat.get("threads", 0)) + stat_changed |= _set_if_changed(cs, "responses", stat.get("responses", 0)) + stat_changed |= _set_if_changed(cs, "replies", stat.get("replies", 0)) + stat_changed |= _set_if_changed(cs, "deleted_threads", stat.get("deleted_threads", 0)) + stat_changed |= _set_if_changed(cs, "deleted_responses", stat.get("deleted_responses", 0)) + stat_changed |= _set_if_changed(cs, "deleted_replies", stat.get("deleted_replies", 0)) + stat_changed |= _set_if_changed(cs, "last_activity_at", last_activity_at) + if stat_changed: + stats_to_update.append(cs) else: stats_to_create.append( CourseStat( @@ -235,7 +265,11 @@ def migrate_users(db: Database[dict[str, Any]], course_id: str) -> None: # --------------------------------------------------------------------------- -def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: +def migrate_content( + db: Database[dict[str, Any]], + course_id: str, + updated_since: datetime | None = None, +) -> None: """ Migrate threads and comments from MongoDB to MySQL. @@ -250,56 +284,70 @@ def migrate_content(db: Database[dict[str, Any]], course_id: str) -> None: 6. Issue a SINGLE MongoDB ``$in`` query for all subscriptions instead of one query per content item. """ - contents = list(db.contents.find({"course_id": course_id}).sort("created_at", 1)) - if not contents: - return + content_query: dict[str, Any] = {"course_id": course_id} + if updated_since is not None: + content_query["updated_at"] = {"$gte": updated_since} + + contents = list(db.contents.find(content_query).sort("created_at", 1)) all_ids_str = [str(c["_id"]) for c in contents] - # Pre-fetch all existing MongoContent rows (one query). - mongo_cache: dict[str, MongoContent] = { - mc.mongo_id: mc for mc in MongoContent.objects.filter(mongo_id__in=all_ids_str) - } + # Pre-fetch existing MongoContent rows for selected content. + mongo_cache: dict[str, MongoContent] = {} + if all_ids_str: + mongo_cache = { + mc.mongo_id: mc + for mc in MongoContent.objects.filter(mongo_id__in=all_ids_str) + } - # Collect every user ID referenced anywhere; fetch them all at once. + # Collect every user ID referenced in selected content; fetch them all at once. user_cache = _build_user_cache(_collect_all_user_ids(contents)) # ContentType objects are cached by Django's framework after the first call. thread_ct = ContentType.objects.get_for_model(CommentThread) comment_ct = ContentType.objects.get_for_model(Comment) - threads_data = [c for c in contents if c["_type"] == "CommentThread"] - comments_data = [c for c in contents if c["_type"] == "Comment"] + if contents: + threads_data = [c for c in contents if c["_type"] == "CommentThread"] + comments_data = [c for c in contents if c["_type"] == "Comment"] - # --- Threads --- - _bulk_migrate_threads(threads_data, mongo_cache, user_cache, thread_ct) - _refresh_mongo_cache(mongo_cache, [str(t["_id"]) for t in threads_data]) + # --- Threads --- + _bulk_migrate_threads(threads_data, mongo_cache, user_cache, thread_ct) + _refresh_mongo_cache(mongo_cache, [str(t["_id"]) for t in threads_data]) - # --- Comments: top-level first, then children (parent must exist first) --- - top_level = [ - c - for c in comments_data - if not c.get("parent_id") or str(c.get("parent_id")) == "None" - ] - child_comments = [ - c - for c in comments_data - if c.get("parent_id") and str(c.get("parent_id")) != "None" - ] - _bulk_migrate_comments(top_level, mongo_cache, user_cache, comment_ct) - _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in top_level]) - _bulk_migrate_comments(child_comments, mongo_cache, user_cache, comment_ct) - _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in child_comments]) - - # --- Metadata: votes, edit history, flaggers --- - _bulk_migrate_votes(contents, mongo_cache, user_cache, thread_ct, comment_ct) - _bulk_migrate_edit_history(contents, mongo_cache, user_cache, thread_ct, comment_ct) - _bulk_migrate_abuse_flaggers( - contents, mongo_cache, user_cache, thread_ct, comment_ct - ) + # --- Comments: top-level first, then children (parent must exist first) --- + top_level = [ + c + for c in comments_data + if not c.get("parent_id") or str(c.get("parent_id")) == "None" + ] + child_comments = [ + c + for c in comments_data + if c.get("parent_id") and str(c.get("parent_id")) != "None" + ] + _bulk_migrate_comments(top_level, mongo_cache, user_cache, comment_ct) + _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in top_level]) + _bulk_migrate_comments(child_comments, mongo_cache, user_cache, comment_ct) + _refresh_mongo_cache(mongo_cache, [str(c["_id"]) for c in child_comments]) + + # --- Metadata: votes, edit history, flaggers --- + _bulk_migrate_votes(contents, mongo_cache, user_cache, thread_ct, comment_ct) + _bulk_migrate_edit_history( + contents, mongo_cache, user_cache, thread_ct, comment_ct + ) + _bulk_migrate_abuse_flaggers( + contents, mongo_cache, user_cache, thread_ct, comment_ct + ) - # --- Subscriptions: ONE MongoDB query for the entire course --- - _bulk_migrate_subscriptions(db, all_ids_str, mongo_cache, user_cache) + # --- Subscriptions: query by course (and optional updated_since) --- + _bulk_migrate_subscriptions( + db, + course_id, + mongo_cache, + user_cache, + updated_since=updated_since, + ) def _refresh_mongo_cache(cache: dict[str, MongoContent], mongo_ids: list[str]) -> None: @@ -389,16 +437,26 @@ def _bulk_migrate_threads( # pylint: disable=too-many-statements if pairs: mongo_ids, objs = zip(*pairs) + obj_list = list(objs) + _course_id = obj_list[0].course_id - # MySQL's bulk_create does not always return PKs (Django feature - # flag can_return_rows_from_bulk_insert may be False). Snapshot - # the current max PK so we can re-fetch created rows afterwards. - # This is safe because migration runs as a single writer. + # Snapshot max PK before bulk_create, then re-fetch by pk range + # narrowed to this course_id so a concurrent insert for a different + # course cannot shift the mongo_id ↔ PK zip alignment. max_pk_before = CommentThread.objects.aggregate(Max("pk"))["pk__max"] or 0 - CommentThread.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) + CommentThread.objects.bulk_create(obj_list, batch_size=BATCH_SIZE) created = list( - CommentThread.objects.filter(pk__gt=max_pk_before).order_by("pk") + CommentThread.objects.filter( + pk__gt=max_pk_before, course_id=_course_id + ).order_by("pk") ) + if len(created) != len(obj_list): + raise RuntimeError( + f"Thread bulk_create count mismatch for course {_course_id}: " + f"submitted {len(obj_list)}, re-fetched {len(created)}. " + "A concurrent writer may have inserted rows between the PK " + "snapshot and re-fetch." + ) mc_rows = [ MongoContent( mongo_id=mid, @@ -447,6 +505,7 @@ def _bulk_migrate_threads( # pylint: disable=too-many-statements mongo_cache[str(t["_id"])].content_object_id for t in existing_data if str(t["_id"]) in mongo_cache + and mongo_cache[str(t["_id"])].content_object_id is not None ] thread_pk_map = {th.pk: th for th in CommentThread.objects.filter(pk__in=pks)} thread_update_fields = [ @@ -477,6 +536,7 @@ def _bulk_migrate_threads( # pylint: disable=too-many-statements thread = thread_pk_map.get(mc.content_object_id) if not thread: continue + deleted_by = ( user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] if t.get("deleted_by") @@ -487,25 +547,44 @@ def _bulk_migrate_threads( # pylint: disable=too-many-statements if t.get("closed_by_id") else None ) - thread.title = get_trunc_title(t.get("title", "")) - thread.body = t["body"] - thread.thread_type = t.get("thread_type", "discussion") - thread.context = t.get("context", "course") - thread.anonymous = t.get("anonymous", False) - thread.anonymous_to_peers = t.get("anonymous_to_peers", False) - thread.closed = t.get("closed", False) - thread.closed_by = closed_by # type: ignore[assignment] - thread.close_reason_code = t.get("close_reason_code") - thread.pinned = t.get("pinned", False) - thread.updated_at = parse_mongo_datetime(t["updated_at"]) # type: ignore[assignment] - thread.last_activity_at = parse_mongo_datetime(t["last_activity_at"]) - thread.commentable_id = t.get("commentable_id") # type: ignore[assignment] - thread.is_spam = t.get("is_spam", False) - thread.is_deleted = t.get("is_deleted", False) - thread.deleted_at = parse_mongo_datetime(t.get("deleted_at")) - thread.deleted_by = deleted_by # type: ignore[assignment] - thread.visible = t.get("visible", True) - to_update.append(thread) + updated_at = parse_mongo_datetime(t["updated_at"]) + last_activity_at = parse_mongo_datetime(t["last_activity_at"]) + deleted_at = parse_mongo_datetime(t.get("deleted_at")) + + # Check each field directly to avoid defining a closure inside the + # loop (which captures variables by reference and confuses linters). + changed_fields: list[str] = [] + for _f, _v in ( + ("title", get_trunc_title(t.get("title", ""))), + ("body", t["body"]), + ("thread_type", t.get("thread_type", "discussion")), + ("context", t.get("context", "course")), + ("anonymous", t.get("anonymous", False)), + ("anonymous_to_peers", t.get("anonymous_to_peers", False)), + ("closed", t.get("closed", False)), + ("closed_by", closed_by), + ("close_reason_code", t.get("close_reason_code")), + ("pinned", t.get("pinned", False)), + ("updated_at", updated_at), + ("last_activity_at", last_activity_at), + ("commentable_id", t.get("commentable_id")), + ("is_spam", t.get("is_spam", False)), + ("is_deleted", t.get("is_deleted", False)), + ("deleted_at", deleted_at), + ("deleted_by", deleted_by), + ("visible", t.get("visible", True)), + ): + if _set_if_changed(thread, _f, _v): + changed_fields.append(_f) + + if changed_fields: + logger.info( + "Updated mapped thread during migration: mongo_id=%s thread_id=%s changed_fields=%s", + str(t["_id"]), + thread.pk, + ",".join(changed_fields), + ) + to_update.append(thread) if to_update: CommentThread.objects.bulk_update( to_update, thread_update_fields, batch_size=BATCH_SIZE @@ -607,12 +686,25 @@ def _bulk_migrate_comments( # pylint: disable=too-many-statements if pairs: mongo_ids, objs = zip(*pairs) + obj_list = list(objs) + _course_id = obj_list[0].course_id - # Same max-PK-before trick as for threads: MySQL may not return - # PKs from bulk_create, so we re-fetch created rows by pk range. + # Narrow re-fetch to this course so a concurrent insert for a + # different course cannot misalign the mongo_id ↔ PK zip. max_pk_before = Comment.objects.aggregate(Max("pk"))["pk__max"] or 0 - Comment.objects.bulk_create(list(objs), batch_size=BATCH_SIZE) - created = list(Comment.objects.filter(pk__gt=max_pk_before).order_by("pk")) + Comment.objects.bulk_create(obj_list, batch_size=BATCH_SIZE) + created = list( + Comment.objects.filter( + pk__gt=max_pk_before, course_id=_course_id + ).order_by("pk") + ) + if len(created) != len(obj_list): + raise RuntimeError( + f"Comment bulk_create count mismatch for course {_course_id}: " + f"submitted {len(obj_list)}, re-fetched {len(created)}. " + "A concurrent writer may have inserted rows between the PK " + "snapshot and re-fetch." + ) mc_rows = [] sort_key_updates: list[Comment] = [] @@ -672,6 +764,7 @@ def _bulk_migrate_comments( # pylint: disable=too-many-statements mongo_cache[str(c["_id"])].content_object_id for c in existing_data if str(c["_id"]) in mongo_cache + and mongo_cache[str(c["_id"])].content_object_id is not None ] comment_pk_map = {cm.pk: cm for cm in Comment.objects.filter(pk__in=pks)} comment_update_fields = [ @@ -700,18 +793,26 @@ def _bulk_migrate_comments( # pylint: disable=too-many-statements if c.get("deleted_by") else None ) - comment.body = c["body"] - comment.anonymous = c.get("anonymous", False) - comment.anonymous_to_peers = c.get("anonymous_to_peers", False) - comment.endorsed = c.get("endorsed", False) - comment.child_count = c.get("child_count", 0) - comment.updated_at = parse_mongo_datetime(c["updated_at"]) # type: ignore[assignment] - comment.is_spam = c.get("is_spam", False) - comment.is_deleted = c.get("is_deleted", False) - comment.deleted_at = parse_mongo_datetime(c.get("deleted_at")) - comment.deleted_by = deleted_by # type: ignore[assignment] - comment.visible = c.get("visible", True) - to_update.append(comment) + updated_at = parse_mongo_datetime(c["updated_at"]) + deleted_at = parse_mongo_datetime(c.get("deleted_at")) + + has_changes = False + has_changes |= _set_if_changed(comment, "body", c["body"]) + has_changes |= _set_if_changed(comment, "anonymous", c.get("anonymous", False)) + has_changes |= _set_if_changed( + comment, "anonymous_to_peers", c.get("anonymous_to_peers", False) + ) + has_changes |= _set_if_changed(comment, "endorsed", c.get("endorsed", False)) + has_changes |= _set_if_changed(comment, "child_count", c.get("child_count", 0)) + has_changes |= _set_if_changed(comment, "updated_at", updated_at) + has_changes |= _set_if_changed(comment, "is_spam", c.get("is_spam", False)) + has_changes |= _set_if_changed(comment, "is_deleted", c.get("is_deleted", False)) + has_changes |= _set_if_changed(comment, "deleted_at", deleted_at) + has_changes |= _set_if_changed(comment, "deleted_by", deleted_by) + has_changes |= _set_if_changed(comment, "visible", c.get("visible", True)) + + if has_changes: + to_update.append(comment) if to_update: Comment.objects.bulk_update( to_update, comment_update_fields, batch_size=BATCH_SIZE @@ -743,26 +844,44 @@ def _bulk_migrate_votes( return obj_ids = {obj_id for _, _, obj_id, _ in candidates} - existing_keys = set( - UserVote.objects.filter(content_object_id__in=obj_ids).values_list( - "user_id", "content_type_id", "content_object_id" + # Fetch existing votes WITH their direction so changed votes (e.g. up → down) + # are detected and corrected, not silently skipped. + existing_vote_map: dict[tuple[int, int, int], int] = { + (row[0], row[1], row[2]): row[3] + for row in UserVote.objects.filter(content_object_id__in=obj_ids).values_list( + "user_id", "content_type_id", "content_object_id", "vote" ) - ) + } + + new_votes: list[UserVote] = [] + changed_votes: list[tuple[int, int, int, int]] = [] # (uid, ct_id, obj_id, new_vote) + seen_vote_keys: set[tuple[int, int, int]] = set(existing_vote_map.keys()) + + for uid, ct_id, obj_id, vote_val in candidates: + key = (uid, ct_id, obj_id) + if key not in existing_vote_map: + if key not in seen_vote_keys: + new_votes.append( + UserVote( + user_id=uid, + content_type_id=ct_id, + content_object_id=obj_id, + vote=vote_val, + ) + ) + seen_vote_keys.add(key) + elif existing_vote_map[key] != vote_val: + # Vote direction changed between Phase 1 and Phase 2 refresh. + changed_votes.append((uid, ct_id, obj_id, vote_val)) - new_votes = [ - UserVote( - user_id=uid, - content_type_id=ct_id, - content_object_id=obj_id, - vote=vote_val, - ) - for uid, ct_id, obj_id, vote_val in candidates - if (uid, ct_id, obj_id) not in existing_keys - ] if new_votes: UserVote.objects.bulk_create( new_votes, batch_size=BATCH_SIZE, ignore_conflicts=True ) + for uid, ct_id, obj_id, vote_val in changed_votes: + UserVote.objects.filter( + user_id=uid, content_type_id=ct_id, content_object_id=obj_id + ).update(vote=vote_val) def _bulk_migrate_edit_history( @@ -888,9 +1007,10 @@ def _bulk_migrate_abuse_flaggers( def _bulk_migrate_subscriptions( db: Database[dict[str, Any]], - content_ids_str: list[str], + course_id: str, mongo_cache: dict[str, MongoContent], user_cache: dict[int, User], + updated_since: datetime | None = None, ) -> None: """ Migrate subscriptions for an entire course using a SINGLE MongoDB query. @@ -898,28 +1018,59 @@ def _bulk_migrate_subscriptions( The original code fired one ``db.subscriptions.find()`` per content item. This version fetches them all at once via ``$in``. """ - if not content_ids_str: - return + sub_query: dict[str, Any] = {"source.course_id": course_id} + if updated_since is not None: + sub_query["updated_at"] = {"$gte": updated_since} - all_subs = list(db.subscriptions.find({"source_id": {"$in": content_ids_str}})) + all_subs = list(db.subscriptions.find(sub_query)) if not all_subs: return thread_ct = ContentType.objects.get_for_model(CommentThread) comment_ct = ContentType.objects.get_for_model(Comment) - obj_ids = [ - mc.content_object_id for mc in mongo_cache.values() if mc.content_object_id + source_ids = { + str(sub.get("source_id")) + for sub in all_subs + if sub.get("source_id") is not None + } + + missing_source_ids = [sid for sid in source_ids if sid not in mongo_cache] + if missing_source_ids: + mongo_cache.update( + { + mc.mongo_id: mc + for mc in MongoContent.objects.filter(mongo_id__in=missing_source_ids) + } + ) + + mapped_source_obj_ids = [ + mc.content_object_id + for sid in source_ids + for mc in [mongo_cache.get(sid)] + if mc and mc.content_object_id ] + existing_sub_map: dict[tuple[int, int, int], Subscription] = { ( cast(int, s.subscriber_id), # type: ignore[attr-defined] cast(int, s.source_content_type_id), # type: ignore[attr-defined] s.source_object_id, ): s - for s in Subscription.objects.filter(source_object_id__in=obj_ids) + for s in Subscription.objects.filter(source_object_id__in=mapped_source_obj_ids) } + # Subscription subscribers who never authored/voted on content are absent + # from the caller's user_cache. Fetch any missing ones now. + missing_sub_uids: set[int] = set() + for sub in all_subs: + uid = _to_int_id(sub.get("subscriber_id")) + if uid is not None and uid not in user_cache: + missing_sub_uids.add(uid) + if missing_sub_uids: + extra_users = {u.pk: u for u in User.objects.filter(pk__in=missing_sub_uids)} + user_cache = {**user_cache, **extra_users} + subs_to_create: list[Subscription] = [] subs_to_update: list[Subscription] = [] seen: set[tuple[int, int, int]] = set(existing_sub_map.keys()) @@ -942,9 +1093,13 @@ def _bulk_migrate_subscriptions( if key in existing_sub_map: s = existing_sub_map[key] - s.created_at = created_at - s.updated_at = updated_at - subs_to_update.append(s) + # updated_at is the reliable change signal for subscriptions: + # it is set on creation and advances on any mutation. If it + # matches what is already in MySQL, nothing changed — skip. + if s.updated_at != updated_at: + s.created_at = created_at + s.updated_at = updated_at + subs_to_update.append(s) elif key not in seen: subs_to_create.append( Subscription( @@ -1069,8 +1224,9 @@ def migrate_read_states(db: Database[dict[str, Any]], course_id: str) -> None: key = (rs.pk, thread_pk) if key in existing_lrt: lrt = existing_lrt[key] - lrt.timestamp = parsed_ts # type: ignore[assignment] - lrt_to_update.append(lrt) + if lrt.timestamp != parsed_ts: + lrt.timestamp = parsed_ts # type: ignore[assignment] + lrt_to_update.append(lrt) elif key not in seen_keys: lrt_to_create.append( LastReadTime( @@ -1098,15 +1254,14 @@ def delete_course_data( stdout: OutputWrapper, ) -> None: """Delete content (threads and comments).""" - contents = db.contents.find({"course_id": course_id}) - for content in contents: - subscriptions = ( - db.subscriptions.delete_many({"source_id": str(content["_id"])}) - if not dry_run - else None - ) + contents = list(db.contents.find({"course_id": course_id})) + content_ids = [str(c["_id"]) for c in contents] + # Delete all subscriptions in one $in query instead of one-per-content. + sub_result = None + if content_ids and not dry_run: + sub_result = db.subscriptions.delete_many({"source_id": {"$in": content_ids}}) stdout.write( - f"Subscription documents to be deleted: {subscriptions.deleted_count if subscriptions else 'N/A (dry run)'}" + f"Subscription documents to be deleted: {sub_result.deleted_count if sub_result else 'N/A (dry run)'}" ) content_result = ( @@ -1163,6 +1318,8 @@ def enable_mysql_backend_for_course(course_id: str) -> None: from forum.toggles import ENABLE_MYSQL_BACKEND course_key = CourseKey.from_string(course_id) - WaffleFlagCourseOverrideModel.objects.create( - course_id=course_key, waffle_flag=ENABLE_MYSQL_BACKEND.name, enabled=True + WaffleFlagCourseOverrideModel.objects.update_or_create( + course_id=course_key, + waffle_flag=ENABLE_MYSQL_BACKEND.name, + defaults={"enabled": True}, ) diff --git a/tests/test_management/test_commands/test_migration_commands.py b/tests/test_management/test_commands/test_migration_commands.py index 06965897..091037ad 100644 --- a/tests/test_management/test_commands/test_migration_commands.py +++ b/tests/test_management/test_commands/test_migration_commands.py @@ -946,3 +946,292 @@ def test_migrate_comment_fallback_to_current_username( mongo_comment = MongoContent.objects.get(mongo_id=comment_id) comment = Comment.objects.get(pk=mongo_comment.content_object_id) assert comment.author_username == "current_username" + + +def test_remigrate_updates_changed_thread_and_comment( + patched_mongodb: Database[Any], +) -> None: + """Ensure remigration updates already-mapped thread/comment when Mongo changed.""" + thread_id = ObjectId() + comment_id = ObjectId() + first_time = timezone.now() + second_time = first_time + timezone.timedelta(minutes=5) + + patched_mongodb.contents.insert_many( + [ + { + "_id": thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Initial Title", + "body": "Initial thread body", + "created_at": first_time, + "updated_at": first_time, + "last_activity_at": first_time, + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + }, + { + "_id": comment_id, + "_type": "Comment", + "author_id": "1", + "course_id": "test_course", + "body": "Initial comment body", + "created_at": first_time, + "updated_at": first_time, + "comment_thread_id": thread_id, + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + "depth": 0, + "sk": f"{comment_id}", + }, + ] + ) + + User.objects.create(id=1, username="testuser") + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + patched_mongodb.contents.update_one( + {"_id": thread_id}, + { + "$set": { + "title": "Updated Title", + "body": "Updated thread body", + "updated_at": second_time, + "last_activity_at": second_time, + } + }, + ) + patched_mongodb.contents.update_one( + {"_id": comment_id}, + { + "$set": { + "body": "Updated comment body", + "updated_at": second_time, + } + }, + ) + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + mongo_thread = MongoContent.objects.get(mongo_id=thread_id) + migrated_thread = CommentThread.objects.get(pk=mongo_thread.content_object_id) + assert migrated_thread.title == "Updated Title" + assert migrated_thread.body == "Updated thread body" + + mongo_comment = MongoContent.objects.get(mongo_id=comment_id) + migrated_comment = Comment.objects.get(pk=mongo_comment.content_object_id) + assert migrated_comment.body == "Updated comment body" + + +def test_remigrate_skips_noop_thread_bulk_update( + patched_mongodb: Database[Any], monkeypatch: pytest.MonkeyPatch +) -> None: + """Ensure unchanged mapped threads are not sent to bulk_update.""" + thread_id = ObjectId() + now = timezone.now() + + patched_mongodb.contents.insert_one( + { + "_id": thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Stable Title", + "body": "Stable body", + "created_at": now, + "updated_at": now, + "last_activity_at": now, + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + } + ) + User.objects.create(id=1, username="testuser") + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + original_bulk_update = CommentThread.objects.bulk_update + bulk_update_calls: list[int] = [] + + def _tracking_bulk_update( + objs: list[CommentThread], + fields: list[str], + batch_size: int | None = None, + ) -> None: + bulk_update_calls.append(len(objs)) + original_bulk_update(objs, fields, batch_size=batch_size) + + monkeypatch.setattr(CommentThread.objects, "bulk_update", _tracking_bulk_update) + + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + assert bulk_update_calls == [] + + +def test_updated_since_filters_content(patched_mongodb: Database[Any]) -> None: + """Only content updated on/after cutoff should be migrated.""" + old_thread_id = ObjectId() + new_thread_id = ObjectId() + now = timezone.now() + cutoff = now - timezone.timedelta(hours=1) + + User.objects.create(id=1, username="testuser") + + patched_mongodb.contents.insert_many( + [ + { + "_id": old_thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Old thread", + "body": "Old body", + "created_at": now - timezone.timedelta(days=2), + "updated_at": now - timezone.timedelta(days=2), + "last_activity_at": now - timezone.timedelta(days=2), + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + }, + { + "_id": new_thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "New thread", + "body": "New body", + "created_at": now - timezone.timedelta(minutes=30), + "updated_at": now - timezone.timedelta(minutes=30), + "last_activity_at": now - timezone.timedelta(minutes=30), + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + }, + ] + ) + + call_command( + "forum_migrate_course_from_mongodb_to_mysql", + "test_course", + "--updated-since", + cutoff.isoformat(), + ) + + assert not MongoContent.objects.filter(mongo_id=str(old_thread_id)).exists() + assert MongoContent.objects.filter(mongo_id=str(new_thread_id)).exists() + + +def test_updated_since_filters_users(patched_mongodb: Database[Any]) -> None: + """Only users with recent course_stats.last_activity_at should be migrated.""" + now = timezone.now() + cutoff = now - timezone.timedelta(hours=1) + + User.objects.create(id=1, username="olduser") + User.objects.create(id=2, username="newuser") + + patched_mongodb.users.insert_many( + [ + { + "_id": "1", + "username": "olduser", + "default_sort_key": "date", + "course_stats": [ + { + "course_id": "test_course", + "threads": 1, + "last_activity_at": now - timezone.timedelta(days=2), + } + ], + }, + { + "_id": "2", + "username": "newuser", + "default_sort_key": "date", + "course_stats": [ + { + "course_id": "test_course", + "threads": 2, + "last_activity_at": now - timezone.timedelta(minutes=10), + } + ], + }, + ] + ) + + call_command( + "forum_migrate_course_from_mongodb_to_mysql", + "test_course", + "--updated-since", + cutoff.isoformat(), + ) + + assert not CourseStat.objects.filter(user_id=1, course_id="test_course").exists() + assert CourseStat.objects.filter(user_id=2, course_id="test_course").exists() + + +def test_updated_since_filters_subscriptions(patched_mongodb: Database[Any]) -> None: + """Subscription delta should use subscription.updated_at, even if content is older.""" + now = timezone.now() + cutoff = now - timezone.timedelta(hours=1) + thread_id = ObjectId() + + User.objects.create(id=1, username="author") + User.objects.create(id=2, username="sub_old") + User.objects.create(id=3, username="sub_new") + + patched_mongodb.contents.insert_one( + { + "_id": thread_id, + "_type": "CommentThread", + "author_id": "1", + "course_id": "test_course", + "title": "Thread", + "body": "Body", + "created_at": now - timezone.timedelta(days=2), + "updated_at": now - timezone.timedelta(days=2), + "last_activity_at": now - timezone.timedelta(days=2), + "votes": {"up": [], "down": []}, + "abuse_flaggers": [], + "historical_abuse_flaggers": [], + } + ) + + # First run creates MongoContent mapping for the thread. + call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") + + patched_mongodb.subscriptions.insert_many( + [ + { + "subscriber_id": "2", + "source_id": str(thread_id), + "source_type": "CommentThread", + "source": {"course_id": "test_course"}, + "created_at": now - timezone.timedelta(days=2), + "updated_at": now - timezone.timedelta(days=2), + }, + { + "subscriber_id": "3", + "source_id": str(thread_id), + "source_type": "CommentThread", + "source": {"course_id": "test_course"}, + "created_at": now - timezone.timedelta(minutes=30), + "updated_at": now - timezone.timedelta(minutes=30), + }, + ] + ) + + call_command( + "forum_migrate_course_from_mongodb_to_mysql", + "test_course", + "--updated-since", + cutoff.isoformat(), + ) + + subs = Subscription.objects.all() + assert subs.count() == 1 + assert subs.first().subscriber_id == 3 # type: ignore[union-attr] From a023694aef02c1bc792108cf61fb7cd083a8ef01 Mon Sep 17 00:00:00 2001 From: Alam-2U Date: Fri, 17 Jul 2026 12:39:52 +0000 Subject: [PATCH 4/4] fix: optimize forum migration script --- ...um_migrate_course_from_mongodb_to_mysql.py | 9 +-- forum/migration_helpers.py | 70 ++++++++++++++----- .../test_commands/test_migration_commands.py | 42 +++++------ 3 files changed, 80 insertions(+), 41 deletions(-) diff --git a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py index 25c86fdb..81d30ac1 100644 --- a/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py +++ b/forum/management/commands/forum_migrate_course_from_mongodb_to_mysql.py @@ -1,9 +1,9 @@ """Migration command for courses from mongodb to mysql.""" -from datetime import datetime +from datetime import datetime, timezone as dt_timezone import time from concurrent.futures import ThreadPoolExecutor, as_completed -from typing import Any +from typing import Any, cast from django.core.management.base import BaseCommand, CommandError, CommandParser from django.db import connections @@ -83,7 +83,7 @@ def _parse_updated_since(raw: str | None) -> datetime | None: ) from exc if timezone.is_naive(parsed): - parsed = timezone.make_aware(parsed, timezone=timezone.utc) + parsed = timezone.make_aware(parsed, timezone=dt_timezone.utc) return parsed def add_arguments(self, parser: CommandParser) -> None: @@ -136,7 +136,8 @@ def handle(self, *args: str, **options: dict[str, Any]) -> None: create_waffle_flags = not options["no_toggle"] workers: int = int(str(options["workers"])) batch_size: int = int(str(options["batch_size"])) - updated_since = self._parse_updated_since(options.get("updated_since")) + updated_since_raw = cast(str | None, options.get("updated_since")) + updated_since = self._parse_updated_since(updated_since_raw) if workers < 1: raise CommandError("--workers must be >= 1.") diff --git a/forum/migration_helpers.py b/forum/migration_helpers.py index 55e21dec..5a4c4063 100644 --- a/forum/migration_helpers.py +++ b/forum/migration_helpers.py @@ -128,7 +128,7 @@ def _set_if_changed(obj: Any, field: str, value: Any) -> bool: # --------------------------------------------------------------------------- -def migrate_users( +def migrate_users( # pylint: disable=too-many-statements db: Database[dict[str, Any]], course_id: str, updated_since: datetime | None = None, @@ -189,7 +189,9 @@ def migrate_users( new_forum_users, batch_size=BATCH_SIZE, ignore_conflicts=True ) if fu_to_update: - ForumUser.objects.bulk_update(fu_to_update, ["default_sort_key"], batch_size=BATCH_SIZE) + ForumUser.objects.bulk_update( + fu_to_update, ["default_sort_key"], batch_size=BATCH_SIZE + ) # --- CourseStat: bulk create new / bulk update existing --- existing_stats: dict[int, CourseStat] = { @@ -222,15 +224,29 @@ def migrate_users( if uid in existing_stats: cs = existing_stats[uid] stat_changed = False - stat_changed |= _set_if_changed(cs, "active_flags", stat.get("active_flags", 0)) - stat_changed |= _set_if_changed(cs, "inactive_flags", stat.get("inactive_flags", 0)) + stat_changed |= _set_if_changed( + cs, "active_flags", stat.get("active_flags", 0) + ) + stat_changed |= _set_if_changed( + cs, "inactive_flags", stat.get("inactive_flags", 0) + ) stat_changed |= _set_if_changed(cs, "threads", stat.get("threads", 0)) - stat_changed |= _set_if_changed(cs, "responses", stat.get("responses", 0)) + stat_changed |= _set_if_changed( + cs, "responses", stat.get("responses", 0) + ) stat_changed |= _set_if_changed(cs, "replies", stat.get("replies", 0)) - stat_changed |= _set_if_changed(cs, "deleted_threads", stat.get("deleted_threads", 0)) - stat_changed |= _set_if_changed(cs, "deleted_responses", stat.get("deleted_responses", 0)) - stat_changed |= _set_if_changed(cs, "deleted_replies", stat.get("deleted_replies", 0)) - stat_changed |= _set_if_changed(cs, "last_activity_at", last_activity_at) + stat_changed |= _set_if_changed( + cs, "deleted_threads", stat.get("deleted_threads", 0) + ) + stat_changed |= _set_if_changed( + cs, "deleted_responses", stat.get("deleted_responses", 0) + ) + stat_changed |= _set_if_changed( + cs, "deleted_replies", stat.get("deleted_replies", 0) + ) + stat_changed |= _set_if_changed( + cs, "last_activity_at", last_activity_at + ) if stat_changed: stats_to_update.append(cs) else: @@ -537,6 +553,14 @@ def _bulk_migrate_threads( # pylint: disable=too-many-statements if not thread: continue + updated_at = parse_mongo_datetime(t["updated_at"]) + + # Fast-skip: if updated_at already matches what is in MySQL this + # thread was fully migrated in a prior run and nothing changed. + # Avoids all field comparisons and the SQL UPDATE on retry. + if thread.updated_at == updated_at: + continue + deleted_by = ( user_cache.get(_to_int_id(t.get("deleted_by"))) # type: ignore[arg-type] if t.get("deleted_by") @@ -547,7 +571,6 @@ def _bulk_migrate_threads( # pylint: disable=too-many-statements if t.get("closed_by_id") else None ) - updated_at = parse_mongo_datetime(t["updated_at"]) last_activity_at = parse_mongo_datetime(t["last_activity_at"]) deleted_at = parse_mongo_datetime(t.get("deleted_at")) @@ -788,25 +811,38 @@ def _bulk_migrate_comments( # pylint: disable=too-many-statements comment: Comment | None = comment_pk_map.get(mc.content_object_id) # type: ignore[no-redef] if not comment: continue + updated_at = parse_mongo_datetime(c["updated_at"]) + + # Fast-skip: if updated_at already matches MySQL, nothing changed. + if comment.updated_at == updated_at: + continue + deleted_by = ( user_cache.get(_to_int_id(c.get("deleted_by"))) # type: ignore[arg-type] if c.get("deleted_by") else None ) - updated_at = parse_mongo_datetime(c["updated_at"]) deleted_at = parse_mongo_datetime(c.get("deleted_at")) has_changes = False has_changes |= _set_if_changed(comment, "body", c["body"]) - has_changes |= _set_if_changed(comment, "anonymous", c.get("anonymous", False)) + has_changes |= _set_if_changed( + comment, "anonymous", c.get("anonymous", False) + ) has_changes |= _set_if_changed( comment, "anonymous_to_peers", c.get("anonymous_to_peers", False) ) - has_changes |= _set_if_changed(comment, "endorsed", c.get("endorsed", False)) - has_changes |= _set_if_changed(comment, "child_count", c.get("child_count", 0)) + has_changes |= _set_if_changed( + comment, "endorsed", c.get("endorsed", False) + ) + has_changes |= _set_if_changed( + comment, "child_count", c.get("child_count", 0) + ) has_changes |= _set_if_changed(comment, "updated_at", updated_at) has_changes |= _set_if_changed(comment, "is_spam", c.get("is_spam", False)) - has_changes |= _set_if_changed(comment, "is_deleted", c.get("is_deleted", False)) + has_changes |= _set_if_changed( + comment, "is_deleted", c.get("is_deleted", False) + ) has_changes |= _set_if_changed(comment, "deleted_at", deleted_at) has_changes |= _set_if_changed(comment, "deleted_by", deleted_by) has_changes |= _set_if_changed(comment, "visible", c.get("visible", True)) @@ -854,7 +890,9 @@ def _bulk_migrate_votes( } new_votes: list[UserVote] = [] - changed_votes: list[tuple[int, int, int, int]] = [] # (uid, ct_id, obj_id, new_vote) + changed_votes: list[tuple[int, int, int, int]] = ( + [] + ) # (uid, ct_id, obj_id, new_vote) seen_vote_keys: set[tuple[int, int, int]] = set(existing_vote_map.keys()) for uid, ct_id, obj_id, vote_val in candidates: diff --git a/tests/test_management/test_commands/test_migration_commands.py b/tests/test_management/test_commands/test_migration_commands.py index 091037ad..3b8b915c 100644 --- a/tests/test_management/test_commands/test_migration_commands.py +++ b/tests/test_management/test_commands/test_migration_commands.py @@ -1,5 +1,6 @@ """Test forum mongodb migration commands.""" +from datetime import timedelta from io import StringIO from typing import Any @@ -23,7 +24,6 @@ ) from forum.utils import get_trunc_title - pytestmark = pytest.mark.django_db @@ -955,7 +955,7 @@ def test_remigrate_updates_changed_thread_and_comment( thread_id = ObjectId() comment_id = ObjectId() first_time = timezone.now() - second_time = first_time + timezone.timedelta(minutes=5) + second_time = first_time + timedelta(minutes=5) patched_mongodb.contents.insert_many( [ @@ -1070,7 +1070,7 @@ def _tracking_bulk_update( call_command("forum_migrate_course_from_mongodb_to_mysql", "test_course") - assert bulk_update_calls == [] + assert not bulk_update_calls def test_updated_since_filters_content(patched_mongodb: Database[Any]) -> None: @@ -1078,7 +1078,7 @@ def test_updated_since_filters_content(patched_mongodb: Database[Any]) -> None: old_thread_id = ObjectId() new_thread_id = ObjectId() now = timezone.now() - cutoff = now - timezone.timedelta(hours=1) + cutoff = now - timedelta(hours=1) User.objects.create(id=1, username="testuser") @@ -1091,9 +1091,9 @@ def test_updated_since_filters_content(patched_mongodb: Database[Any]) -> None: "course_id": "test_course", "title": "Old thread", "body": "Old body", - "created_at": now - timezone.timedelta(days=2), - "updated_at": now - timezone.timedelta(days=2), - "last_activity_at": now - timezone.timedelta(days=2), + "created_at": now - timedelta(days=2), + "updated_at": now - timedelta(days=2), + "last_activity_at": now - timedelta(days=2), "votes": {"up": [], "down": []}, "abuse_flaggers": [], "historical_abuse_flaggers": [], @@ -1105,9 +1105,9 @@ def test_updated_since_filters_content(patched_mongodb: Database[Any]) -> None: "course_id": "test_course", "title": "New thread", "body": "New body", - "created_at": now - timezone.timedelta(minutes=30), - "updated_at": now - timezone.timedelta(minutes=30), - "last_activity_at": now - timezone.timedelta(minutes=30), + "created_at": now - timedelta(minutes=30), + "updated_at": now - timedelta(minutes=30), + "last_activity_at": now - timedelta(minutes=30), "votes": {"up": [], "down": []}, "abuse_flaggers": [], "historical_abuse_flaggers": [], @@ -1129,7 +1129,7 @@ def test_updated_since_filters_content(patched_mongodb: Database[Any]) -> None: def test_updated_since_filters_users(patched_mongodb: Database[Any]) -> None: """Only users with recent course_stats.last_activity_at should be migrated.""" now = timezone.now() - cutoff = now - timezone.timedelta(hours=1) + cutoff = now - timedelta(hours=1) User.objects.create(id=1, username="olduser") User.objects.create(id=2, username="newuser") @@ -1144,7 +1144,7 @@ def test_updated_since_filters_users(patched_mongodb: Database[Any]) -> None: { "course_id": "test_course", "threads": 1, - "last_activity_at": now - timezone.timedelta(days=2), + "last_activity_at": now - timedelta(days=2), } ], }, @@ -1156,7 +1156,7 @@ def test_updated_since_filters_users(patched_mongodb: Database[Any]) -> None: { "course_id": "test_course", "threads": 2, - "last_activity_at": now - timezone.timedelta(minutes=10), + "last_activity_at": now - timedelta(minutes=10), } ], }, @@ -1177,7 +1177,7 @@ def test_updated_since_filters_users(patched_mongodb: Database[Any]) -> None: def test_updated_since_filters_subscriptions(patched_mongodb: Database[Any]) -> None: """Subscription delta should use subscription.updated_at, even if content is older.""" now = timezone.now() - cutoff = now - timezone.timedelta(hours=1) + cutoff = now - timedelta(hours=1) thread_id = ObjectId() User.objects.create(id=1, username="author") @@ -1192,9 +1192,9 @@ def test_updated_since_filters_subscriptions(patched_mongodb: Database[Any]) -> "course_id": "test_course", "title": "Thread", "body": "Body", - "created_at": now - timezone.timedelta(days=2), - "updated_at": now - timezone.timedelta(days=2), - "last_activity_at": now - timezone.timedelta(days=2), + "created_at": now - timedelta(days=2), + "updated_at": now - timedelta(days=2), + "last_activity_at": now - timedelta(days=2), "votes": {"up": [], "down": []}, "abuse_flaggers": [], "historical_abuse_flaggers": [], @@ -1211,16 +1211,16 @@ def test_updated_since_filters_subscriptions(patched_mongodb: Database[Any]) -> "source_id": str(thread_id), "source_type": "CommentThread", "source": {"course_id": "test_course"}, - "created_at": now - timezone.timedelta(days=2), - "updated_at": now - timezone.timedelta(days=2), + "created_at": now - timedelta(days=2), + "updated_at": now - timedelta(days=2), }, { "subscriber_id": "3", "source_id": str(thread_id), "source_type": "CommentThread", "source": {"course_id": "test_course"}, - "created_at": now - timezone.timedelta(minutes=30), - "updated_at": now - timezone.timedelta(minutes=30), + "created_at": now - timedelta(minutes=30), + "updated_at": now - timedelta(minutes=30), }, ] )