From 507495ac0fc1e20040772e82059921fec3d6ba0f Mon Sep 17 00:00:00 2001 From: Larry Rios Date: Wed, 11 Nov 2026 09:42:59 -0500 Subject: [PATCH] Use correlated EXISTS membership lookups and add relative_path index Check repository version membership per matching ContentArtifact row with a correlated EXISTS over the RepositoryContent version interval, instead of content__in=repo_version.content (unnest of content_ids). This covers the content handler's directory listing, index.html probe, exact file lookup and pass-through file lookup. The index.html probe is existence-only (pk, LIMIT 1) and always runs, including for dotted final path segments, which can be directories. Add a text_pattern_ops btree index on ContentArtifact.relative_path, created concurrently, for the handler's prefix and equality lookups. The migration reuses an existing index only when it is valid and has the expected definition. Tracked in Jira as PULP-2346. Validated in a production A/B test. Co-Authored-By: Claude Sonnet 5.5 --- ...optimize-content-membership-queries.bugfix | 1 + ...entartifact_relative_path_pattern_index.py | 100 +++++++ pulpcore/app/models/content.py | 10 +- pulpcore/content/handler.py | 45 ++- pulpcore/tests/unit/content/test_handler.py | 6 +- .../unit/content/test_handler_membership.py | 268 ++++++++++++++++++ .../test_relative_path_index_migration.py | 119 ++++++++ 7 files changed, 537 insertions(+), 12 deletions(-) create mode 100644 CHANGES/+optimize-content-membership-queries.bugfix create mode 100644 pulpcore/app/migrations/0162_contentartifact_relative_path_pattern_index.py create mode 100644 pulpcore/tests/unit/content/test_handler_membership.py create mode 100644 pulpcore/tests/unit/test_relative_path_index_migration.py diff --git a/CHANGES/+optimize-content-membership-queries.bugfix b/CHANGES/+optimize-content-membership-queries.bugfix new file mode 100644 index 0000000000..b2bdeb9739 --- /dev/null +++ b/CHANGES/+optimize-content-membership-queries.bugfix @@ -0,0 +1 @@ +Sped up content app lookups (the `index.html` probe, directory listings and exact file lookups) by checking repository membership per matching row instead of across the whole repository version, backed by a new index on `ContentArtifact.relative_path`. diff --git a/pulpcore/app/migrations/0162_contentartifact_relative_path_pattern_index.py b/pulpcore/app/migrations/0162_contentartifact_relative_path_pattern_index.py new file mode 100644 index 0000000000..d25136f8b5 --- /dev/null +++ b/pulpcore/app/migrations/0162_contentartifact_relative_path_pattern_index.py @@ -0,0 +1,100 @@ +import re + +import django.contrib.postgres.indexes +from django.db import migrations, models + +_INDEX_NAME = "core_ca_relpath_pattern_idx" +_INDEX_MARKER = "pulpcore:0162:core_ca_relpath_pattern_idx" +_CREATE_INDEX_SQL = ( + 'CREATE INDEX CONCURRENTLY IF NOT EXISTS "core_ca_relpath_pattern_idx" ' + 'ON "core_contentartifact" USING btree ("relative_path" text_pattern_ops)' +) +_DROP_INDEX_SQL = 'DROP INDEX CONCURRENTLY IF EXISTS "core_ca_relpath_pattern_idx"' +# pg_get_indexdef output for the expected index, e.g. +# CREATE INDEX core_ca_relpath_pattern_idx ON public.core_contentartifact USING btree +# (relative_path text_pattern_ops) +_EXPECTED_INDEX_DEF = re.compile( + r'^CREATE INDEX "?core_ca_relpath_pattern_idx"? ON (?:"?\w+"?\.)?"?core_contentartifact"? ' + r'USING btree \("?relative_path"? text_pattern_ops\)$' +) +_COMMENT_INDEX_SQL = ( + 'COMMENT ON INDEX "core_ca_relpath_pattern_idx" ' + "IS 'pulpcore:0162:core_ca_relpath_pattern_idx'" +) + + +def _index_info(cursor): + cursor.execute( + """ + SELECT + pg_index.indisvalid, + obj_description(indexrel.oid, 'pg_class'), + pg_get_indexdef(indexrel.oid) + FROM pg_class AS indexrel + JOIN pg_index ON pg_index.indexrelid = indexrel.oid + JOIN pg_namespace AS namespace ON namespace.oid = indexrel.relnamespace + WHERE indexrel.relname = %s AND namespace.nspname = current_schema() + """, + [_INDEX_NAME], + ) + return cursor.fetchone() + + +def create_relative_path_index(apps, schema_editor): + """Create the index, reusing an equivalent one that already exists (e.g. created out of band).""" + with schema_editor.connection.cursor() as cursor: + info = _index_info(cursor) + if info: + valid, _, definition = info + if not valid: + raise RuntimeError( + f"Index {_INDEX_NAME!r} exists but is invalid; refusing to drop it automatically." + ) + if not _EXPECTED_INDEX_DEF.match(definition): + raise RuntimeError( + f"Index {_INDEX_NAME!r} exists with an unexpected definition ({definition!r}); " + "refusing to reuse or drop it automatically." + ) + return + + # IF NOT EXISTS keeps concurrent migration runs from creating a second physical index. + cursor.execute(_CREATE_INDEX_SQL) + # The comment records ownership so the reverse migration only drops what it created. + cursor.execute(_COMMENT_INDEX_SQL) + + +def drop_relative_path_index(apps, schema_editor): + """Drop only an index created by this migration, leaving pre-existing ones intact.""" + with schema_editor.connection.cursor() as cursor: + info = _index_info(cursor) + if info and info[1] == _INDEX_MARKER: + cursor.execute(_DROP_INDEX_SQL) + + +class Migration(migrations.Migration): + """Add the index used by the content handler's relative_path prefix and equality lookups.""" + + atomic = False # CREATE/DROP INDEX CONCURRENTLY cannot run in a transaction. + + dependencies = [ + ("core", "0161_upstreampulp_remote_policy"), + ] + + operations = [ + migrations.SeparateDatabaseAndState( + state_operations=[ + migrations.AddIndex( + model_name="contentartifact", + index=models.Index( + django.contrib.postgres.indexes.OpClass( + "relative_path", name="text_pattern_ops" + ), + name=_INDEX_NAME, + ), + ), + ], + database_operations=[ + migrations.RunPython(create_relative_path_index, drop_relative_path_index), + ], + ), + ] diff --git a/pulpcore/app/models/content.py b/pulpcore/app/models/content.py index dc9ef3fecd..0518efd3e2 100644 --- a/pulpcore/app/models/content.py +++ b/pulpcore/app/models/content.py @@ -16,7 +16,7 @@ from django.conf import settings from django.contrib.postgres.fields import HStoreField -from django.contrib.postgres.indexes import GinIndex +from django.contrib.postgres.indexes import GinIndex, OpClass from django.core import validators from django.db import IntegrityError, models, transaction from django.forms.models import model_to_dict @@ -663,7 +663,13 @@ class ContentArtifact(BaseModel, QueryMixin): class Meta: unique_together = ("content", "relative_path") - indexes = [models.Index(fields=["relative_path"])] + indexes = [ + models.Index(fields=["relative_path"]), + models.Index( + OpClass("relative_path", name="text_pattern_ops"), + name="core_ca_relpath_pattern_idx", + ), + ] @staticmethod def sort_key(ca): diff --git a/pulpcore/content/handler.py b/pulpcore/content/handler.py index 1bf9104960..dfef3a4be7 100644 --- a/pulpcore/content/handler.py +++ b/pulpcore/content/handler.py @@ -46,6 +46,7 @@ models, transaction, ) +from django.db.models import Exists, OuterRef # noqa: E402 from jinja2 import Template # noqa: E402 from pulpcore.app import mime_types # noqa: E402 @@ -712,7 +713,12 @@ def list_directory_blocking(): if repo_version or publication.pass_through: cas = ContentArtifact.objects.select_related("artifact").filter( - content__in=content_repo_ver.content, relative_path__startswith=path + Exists( + content_repo_ver._content_relationships().filter( + content_id=OuterRef("content_id") + ) + ), + relative_path__startswith=path, ) for ca in cas: name = file_or_directory_name(path, ca.relative_path) @@ -898,9 +904,14 @@ async def _match_and_stream(self, path, request): "artifact", "artifact__pulp_domain" ) .filter( - content__in=publication.repository_version.content, + Exists( + publication.repository_version._content_relationships().filter( + content_id=OuterRef("content_id") + ) + ), + relative_path=original_rel_path, ) - .aget(relative_path=original_rel_path) + .aget() ) except MultipleObjectsReturned: @@ -938,9 +949,14 @@ async def _match_and_stream(self, path, request): # Look for index.html or list the directory index_path = "{}index.html".format(rel_path) - contentartifact_exists = await ContentArtifact.objects.filter( - content__in=repo_version.content, relative_path=index_path - ).aexists() + # Existence-only probe: select just the pk and stop at the first match. + index_matches = ContentArtifact.objects.filter( + Exists( + repo_version._content_relationships().filter(content_id=OuterRef("content_id")) + ), + relative_path=index_path, + ).values_list("pk", flat=True)[:1] + contentartifact_exists = bool([pk async for pk in index_matches]) if contentartifact_exists: original_rel_path = index_path headers = self.response_headers(original_rel_path, distro) @@ -961,9 +977,20 @@ async def _match_and_stream(self, path, request): ) try: - ca = await ContentArtifact.objects.select_related( - "artifact", "artifact__pulp_domain" - ).aget(content__in=repo_version.content, relative_path=original_rel_path) + ca = ( + await ContentArtifact.objects.select_related( + "artifact", "artifact__pulp_domain" + ) + .filter( + Exists( + repo_version._content_relationships().filter( + content_id=OuterRef("content_id") + ) + ), + relative_path=original_rel_path, + ) + .aget() + ) except MultipleObjectsReturned: log.error( diff --git a/pulpcore/tests/unit/content/test_handler.py b/pulpcore/tests/unit/content/test_handler.py index a720bedc3a..4cd65cdcb8 100644 --- a/pulpcore/tests/unit/content/test_handler.py +++ b/pulpcore/tests/unit/content/test_handler.py @@ -308,7 +308,11 @@ async def test_list_directory_metadata(monkeypatch, use_publication): SimpleNamespace(objects=remote_artifact_manager), ) - repo_version = SimpleNamespace(content=[], _content_relationships=lambda: []) + class Relationships(list): + def filter(self, *args, **kwargs): + return Mock() + + repo_version = SimpleNamespace(content=[], _content_relationships=lambda: Relationships()) if use_publication: publication_manager = Mock() diff --git a/pulpcore/tests/unit/content/test_handler_membership.py b/pulpcore/tests/unit/content/test_handler_membership.py new file mode 100644 index 0000000000..e818d5f2ae --- /dev/null +++ b/pulpcore/tests/unit/content/test_handler_membership.py @@ -0,0 +1,268 @@ +"""Database-backed checks of the content handler's repository membership lookups.""" + +import uuid +from unittest.mock import AsyncMock + +import pytest +import pytest_asyncio +from aiohttp.test_utils import make_mocked_request +from aiohttp.web_exceptions import HTTPMovedPermanently +from asgiref.sync import sync_to_async +from django.db import connection +from django.test.utils import CaptureQueriesContext + +from pulpcore.content.handler import Handler, PathNotResolved, _current_distribution +from pulpcore.plugin.models import Content, ContentArtifact, Distribution, Repository + +pytestmark = [pytest.mark.asyncio, pytest.mark.django_db] + + +def _delete_tracked(tracked): + # Distributions first, then repositories (cascading versions and memberships), then content. + Distribution.objects.filter(pk__in=tracked["distributions"]).delete() + Repository.objects.filter(pk__in=tracked["repositories"]).delete() + Content.objects.filter(pk__in=tracked["contents"]).delete() + + +@pytest_asyncio.fixture +async def created(db): + """ + Track the rows a test creates and delete exactly those afterwards. + + These tests build data through worker threads, so it is committed and not rolled back with the + test transaction. Left behind, it changes row order and contents for later tests. + """ + tracked = {"distributions": [], "repositories": [], "contents": []} + yield tracked + await sync_to_async(_delete_tracked)(tracked) + + +@pytest.fixture +def build_repository(created): + @sync_to_async + def _build_repository(*steps): + """ + Create a repository with one version per step; return (repository, versions, contents). + + Each step is (paths_to_add, paths_to_remove). Versions are built through new_version() so + both the content_ids array and the RepositoryContent intervals describe the membership. + """ + repository = Repository.objects.create(name=str(uuid.uuid4())) + created["repositories"].append(repository.pk) + repository.CONTENT_TYPES = [Content] + contents = {} + versions = [] + for add, remove in steps: + for path in add: + content = Content.objects.create(pulp_type="core.content") + created["contents"].append(content.pk) + ContentArtifact.objects.create(content=content, relative_path=path) + contents[path] = content + with repository.new_version() as version: + version.add_content(Content.objects.filter(pk__in=[contents[p].pk for p in add])) + version.remove_content( + Content.objects.filter(pk__in=[contents[p].pk for p in remove]) + ) + versions.append(version) + return repository, versions, contents + + return _build_repository + + +@pytest.fixture +def create_distribution(created): + async def _create_distribution(repo_version): + name = str(uuid.uuid4()) + distribution = await Distribution.objects.acreate( + name=name, base_path=name, repository_version=repo_version + ) + created["distributions"].append(distribution.pk) + return distribution + + return _create_distribution + + +@pytest.fixture +def serve(monkeypatch): + """Serve a path from a distribution; return the ContentArtifact the handler picked.""" + + async def _serve(distribution, path): + served = AsyncMock(side_effect=lambda *args: next(a for a in args if hasattr(a, "pk"))) + monkeypatch.setattr(Handler, "_serve_content_artifact", served) + monkeypatch.setattr(Handler, "_stream_content_artifact", served) + monkeypatch.setattr(Handler, "_set_last_modified_header", AsyncMock()) + _current_distribution.set(distribution) + full_path = f"{distribution.base_path}/{path}" + request = make_mocked_request("GET", f"/pulp/content/{full_path}") + return await Handler()._match_and_stream(full_path, request) + + return _serve + + +async def capture_sql(awaitable): + """Capture SQL run by the ORM's worker thread while awaiting awaitable.""" + context = CaptureQueriesContext(connection) + await sync_to_async(context.__enter__)() + try: + result = await awaitable + finally: + await sync_to_async(context.__exit__)(None, None, None) + # captured_queries reads the calling thread's connection, so read it in the worker thread. + queries = await sync_to_async(lambda: [q["sql"] for q in context.captured_queries])() + return result, queries + + +def contentartifact_sql(queries): + return [sql for sql in queries if 'FROM "core_contentartifact"' in sql] + + +async def test_list_directory_lists_only_version_content(build_repository, created): + """Entries come only from the listed version, not from other repositories or no repository.""" + _, (version,), _ = await build_repository( + (["org/a/a-1.0.jar", "org/b/index.html", "top.txt"], []) + ) + _, (other_version,), _ = await build_repository((["org/c/c-1.0.jar", "org/a/a-2.0.jar"], [])) + # Exists as a ContentArtifact, but belongs to no repository version. + orphan = await Content.objects.acreate(pulp_type="core.content") + created["contents"].append(orphan.pk) + await ContentArtifact.objects.acreate(content=orphan, relative_path="org/z.jar") + + directory_list, _, _ = await Handler().list_directory(version, None, "org/") + assert directory_list == {"a/", "b/"} + + other_list, _, _ = await Handler().list_directory(other_version, None, "org/") + assert other_list == {"c/", "a/"} + + +async def test_list_directory_excludes_removed_content(build_repository): + """Content removed in a later version is listed only for the versions that contained it.""" + _, (version_1, version_2), _ = await build_repository( + (["pkg/keep.rpm", "pkg/gone.rpm"], []), + (["pkg/new.rpm"], ["pkg/gone.rpm"]), + ) + + listed_1, _, _ = await Handler().list_directory(version_1, None, "pkg/") + listed_2, _, _ = await Handler().list_directory(version_2, None, "pkg/") + + assert listed_1 == {"keep.rpm", "gone.rpm"} + assert listed_2 == {"keep.rpm", "new.rpm"} + + +async def test_list_directory_lists_dotted_directories(build_repository): + _, (version,), _ = await build_repository( + (["12.1.12/index.html", "12.1.12/a.txt", "top.txt"], []) + ) + + root, _, _ = await Handler().list_directory(version, None, "") + nested, _, _ = await Handler().list_directory(version, None, "12.1.12/") + + assert root == {"12.1.12/", "top.txt"} + assert nested == {"index.html", "a.txt"} + + +async def test_list_directory_query_uses_correlated_exists(build_repository): + _, (version,), _ = await build_repository((["org/a/a-1.0.jar"], [])) + + _, queries = await capture_sql(Handler().list_directory(version, None, "org/")) + + (sql,) = contentartifact_sql(queries) + assert "EXISTS" in sql + assert "unnest" not in sql.lower() + assert "content_ids" not in sql + + +async def test_file_lookup_serves_only_version_content( + serve, build_repository, create_distribution +): + _, (version,), _ = await build_repository((["org/a.jar"], [])) + await build_repository((["org/other.jar"], [])) + distribution = await create_distribution(version) + + served = await serve(distribution, "org/a.jar") + assert served.relative_path == "org/a.jar" + + # Another repository's content is never served, even though the ContentArtifact exists. + with pytest.raises(PathNotResolved): + await serve(distribution, "org/other.jar") + + +async def test_file_lookup_does_not_serve_removed_content( + serve, build_repository, create_distribution +): + _, (version_1, version_2), _ = await build_repository( + (["pkg/gone.rpm"], []), + ([], ["pkg/gone.rpm"]), + ) + + served = await serve(await create_distribution(version_1), "pkg/gone.rpm") + assert served.relative_path == "pkg/gone.rpm" + with pytest.raises(PathNotResolved): + await serve(await create_distribution(version_2), "pkg/gone.rpm") + + +async def test_file_lookup_query_uses_correlated_exists( + serve, build_repository, create_distribution +): + _, (version,), _ = await build_repository((["org/a.jar"], [])) + distribution = await create_distribution(version) + + _, queries = await capture_sql(serve(distribution, "org/a.jar")) + + lookups = contentartifact_sql(queries) + assert lookups + for sql in lookups: + assert "EXISTS" in sql + assert "unnest" not in sql.lower() + assert "content_ids" not in sql + + +@pytest.mark.parametrize("path", ["12.1.12/", "12.1.12"]) +async def test_index_html_is_served_for_dotted_directory( + serve, build_repository, create_distribution, path +): + """A dotted final segment can still be a directory holding index.html.""" + _, (version,), _ = await build_repository((["12.1.12/index.html", "12.1.12/a.txt"], [])) + distribution = await create_distribution(version) + + served = await serve(distribution, path) + + assert served.relative_path == "12.1.12/index.html" + + +async def test_dotted_directory_without_index_redirects_to_listing( + serve, build_repository, create_distribution +): + _, (version,), _ = await build_repository((["12.1.12/a.txt"], [])) + distribution = await create_distribution(version) + + with pytest.raises(HTTPMovedPermanently): + await serve(distribution, "12.1.12") + + +async def test_index_probe_ignores_other_repository_index( + serve, build_repository, create_distribution +): + _, (version,), _ = await build_repository((["docs/page.html"], [])) + await build_repository((["docs/index.html"], [])) + distribution = await create_distribution(version) + + with pytest.raises(HTTPMovedPermanently): + # No index.html in this version, so the directory is listed instead. + await serve(distribution, "docs") + + +async def test_index_probe_is_existence_only_with_limit( + serve, build_repository, create_distribution +): + _, (version,), _ = await build_repository((["docs/index.html"], [])) + distribution = await create_distribution(version) + + _, queries = await capture_sql(serve(distribution, "docs/")) + + probe = next( + sql for sql in contentartifact_sql(queries) if "relative_path\" = 'docs/index.html'" in sql + ) + assert probe.startswith('SELECT "core_contentartifact"."pulp_id"') + assert probe.rstrip().endswith("LIMIT 1") + assert "EXISTS" in probe + assert "unnest" not in probe.lower() diff --git a/pulpcore/tests/unit/test_relative_path_index_migration.py b/pulpcore/tests/unit/test_relative_path_index_migration.py new file mode 100644 index 0000000000..9aa05a95c5 --- /dev/null +++ b/pulpcore/tests/unit/test_relative_path_index_migration.py @@ -0,0 +1,119 @@ +"""Tests for the core_ca_relpath_pattern_idx migration helpers.""" + +import importlib +from types import SimpleNamespace + +import pytest +from django.db import connection + +migration = importlib.import_module( + "pulpcore.app.migrations.0162_contentartifact_relative_path_pattern_index" +) + +INDEX = "core_ca_relpath_pattern_idx" +schema_editor = SimpleNamespace(connection=connection) + + +def index_info(): + with connection.cursor() as cursor: + return migration._index_info(cursor) + + +def replace_index(create_sql, comment=None): + with connection.cursor() as cursor: + cursor.execute(f'DROP INDEX IF EXISTS "{INDEX}"') + cursor.execute(create_sql) + if comment: + cursor.execute(f"COMMENT ON INDEX \"{INDEX}\" IS '{comment}'") + + +@pytest.mark.django_db +def test_equivalent_existing_index_is_reused(): + # The index exists with the expected definition because the test database ran the migration. + before = index_info() + assert migration._EXPECTED_INDEX_DEF.match(before[2]) + + migration.create_relative_path_index(None, schema_editor) + + assert index_info() == before + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "create_sql", + [ + # Default operator class instead of text_pattern_ops. + f'CREATE INDEX "{INDEX}" ON core_contentartifact (relative_path)', + # Wrong column. + f'CREATE INDEX "{INDEX}" ON core_contentartifact (relative_path text_pattern_ops, content_id)', + # Partial index. + f'CREATE INDEX "{INDEX}" ON core_contentartifact (relative_path text_pattern_ops)' + " WHERE relative_path <> ''", + ], + ids=["default-opclass", "extra-column", "partial"], +) +def test_index_with_unexpected_definition_is_rejected(create_sql): + replace_index(create_sql) + before = index_info() + + with pytest.raises(RuntimeError, match="unexpected definition"): + migration.create_relative_path_index(None, schema_editor) + + # The mismatching index is left in place, not dropped or replaced. + assert index_info() == before + + +@pytest.mark.django_db +def test_reverse_leaves_index_it_did_not_create(): + replace_index( + f'CREATE INDEX "{INDEX}" ON core_contentartifact USING btree' + " (relative_path text_pattern_ops)", + comment="created-elsewhere", + ) + + migration.create_relative_path_index(None, schema_editor) + migration.drop_relative_path_index(None, schema_editor) + + assert index_info() is not None + + +class RecordingCursor: + """A stand-in cursor that returns one canned index row and records every statement.""" + + def __init__(self, row): + self.row = row + self.statements = [] + + def __enter__(self): + return self + + def __exit__(self, *exc_info): + return False + + def execute(self, sql, params=None): + self.statements.append(sql) + + def fetchone(self): + return self.row + + +def test_unique_index_is_rejected_without_running_ddl(): + """ + A UNIQUE index of the same name is rejected and left alone. + + It cannot be created in the test database, where relative_path values are not unique, so the + existing index is simulated instead of built. + """ + definition = ( + f"CREATE UNIQUE INDEX {INDEX} ON public.core_contentartifact " + "USING btree (relative_path text_pattern_ops)" + ) + cursor = RecordingCursor((True, None, definition)) + fake_editor = SimpleNamespace(connection=SimpleNamespace(cursor=lambda: cursor)) + + with pytest.raises(RuntimeError, match="unexpected definition"): + migration.create_relative_path_index(None, fake_editor) + + # Only the lookup ran: nothing was created, commented on, or dropped. + assert len(cursor.statements) == 1 + assert cursor.statements[0].lstrip().startswith("SELECT")