Repository: PrefectHQ/prefect
Pull request: Update event occurred + related resource indexes for Postgres
This PR replaces the current `ix_events__event_related_occurred` and `ix_events__related_resource_ids` indexes with new versions that use an `md5` hash of potentially large columns. This fixes an issue where events will many related resources could not be inserted into the database.

I also updated the default value for the maximum number of resources to `100` to match the observed upper limit for SQLite. It can be raised for servers running with a Postgres instance if needed.

Closes https://github.com/PrefectHQ/prefect/issues/17728


Problem statement:
Events with large related_resources causes looping insert error
### Bug summary

When submitting events to prefect the with a large amount of data in the `related` field can cause the `ix_events__event_related_occurred` index can prevent events from being inserted into the database and cause looping errors as Prefect continues to attempt to insert the too large event (I don't know if there's a limit to the number of tries here).

A sample script to highlight this behaviour:

```python
import os
import uuid
from typing import Any

from prefect.events import emit_event


def define_event() -> dict[str, Any]:
    settings_related = os.environ.get("RESOURCE_MULTIPLIER")
    settings_data = os.environ.get("DATA_MULTIPLIER")
    data_multiplier = int(settings_data) if settings_data else 1
    resource_multiplier = int(settings_related) if settings_related else 1

    related_resources = [
        {
            "prefect.resource.id": str(uuid.uuid4()),
            "prefect.resource.role": "test.related",
            "data": str(uuid.uuid4()) * data_multiplier,
        }
        for _ in range(resource_multiplier)
    ]
    event = {
        "event": "test.event",
        "resource": {"prefect.resource.id": str(uuid.uuid4())},
        "related": related_resources,
    }

    print(f"Data multiplier: {data_multiplier}")
    print(f"Related resources: {resource_multiplier}")

    return event


event = define_event()

emit_event(**event)
```

Running this with:

```bash
RESOURCE_MULTIPLIER=35 python emit_events.py
```

Would cause the error loop to begin as this can be seen within the logs of Postgres:

```bash
2025-04-04 09:04:59.614 UTC [129] ERROR:  index row size 2912 exceeds btree version 4 maximum 2704 for index "ix_events__event_related_occurred"
2025-04-04 09:04:59.614 UTC [129] DETAIL:  Index row references tuple (0,4) in relation "events".
2025-04-04 09:04:59.614 UTC [129] HINT:  Values larger than 1/3 of a buffer page cannot be indexed.
        Consider a function index of an MD5 hash of the value, or use full text indexing.
2025-04-04 09:04:59.614 UTC [129] STATEMENT:  INSERT INTO events (occurred, event, resource_id, resource, related_resource_ids, related, payload, received, recorded, follows, id, created, updated) VALUES ($1::TIMESTAMP WITH TIME ZONE, $2::VARCHAR, $3::VARCHAR, $4::JSONB, $5::JSONB, $6::JSONB, $7::JSONB, $8::TIMESTAMP WITH TIME ZONE, $9::TIMESTAMP WITH TIME ZONE, $10::UUID, $11::UUID, $12::TIMESTAMP WITH TIME ZONE, $13::TIMESTAMP WITH TIME ZONE) ON CONFLICT DO NOTHING RETURNING events.id
2025-04-04 09:05:04.620 UTC [129] ERROR:  index row size 2912 exceeds btree version 4 maximum 2704 for index "ix_events__event_related_occurred"
2025-04-04 09:05:04.620 UTC [129] DETAIL:  Index row references tuple (0,5) in relation "events".
2025-04-04 09:05:04.620 UTC [129] HINT:  Values larger than 1/3 of a buffer page cannot be indexed.
        Consider a function index of an MD5 hash of the value, or use full text indexing.
2025-04-04 09:05:04.620 UTC [129] STATEMENT:  INSERT INTO events (occurred, event, resource_id, resource, related_resource_ids, related, payload, received, recorded, follows, id, created, updated) VALUES ($1::TIMESTAMP WITH TIME ZONE, $2::VARCHAR, $3::VARCHAR, $4::JSONB, $5::JSONB, $6::JSONB, $7::JSONB, $8::TIMESTAMP WITH TIME ZONE, $9::TIMESTAMP WITH TIME ZONE, $10::UUID, $11::UUID, $12::TIMESTAMP WITH TIME ZONE, $13::TIMESTAMP WITH TIME ZONE) ON CONFLICT DO NOTHING RETURNING events.id
```

Alternatively if you send in a single related resource with large content.

```bash
DATA_MULTIPLIER=10000 python emit_events.py

```

If the intention is that these type of events shouldn't be supported I'd expect them to be rejected by the Prefect API, otherwise that the index would be swapped for something else.


### Version info

```Text
❯ prefect version
Version:             3.3.2
API version:         0.8.4
Python version:      3.12.8
Git commit:          e49e3185
Built:               Thu, Apr 03, 2025 09:25 PM
OS/Arch:             darwin/arm64
Profile:             ephemeral
Server type:         server
Pydantic version:    2.10.6
Integrations:
  prefect-redis:     0.2.2
```

### Additional context

_No response_

Hints from issue discussion:
Thanks for the issue @ogenstad! There's a setting that controls the maximum number of related resources accepted by the API (`server.events.maxiumum_related_resources`), but it looks like that defaults to 500, so I'm surprised that you're hitting this error so far below that value.

I'm going to try the suggestion in the error of using an MD5 hash in the index and see if that fixes the problem.

Also, because I'm curious, how did you run into this error? Are you emitting custom events with a lot of related resources?
Yes we are generating custom events where some of them can have quite a high number of related resources.

I don't completely know how Postgres converts this internally but based on what I've seen it's more about the size of the payload and how it can be compressed.

So this will work:

```python
from prefect.events import emit_event

emit_event(
    event="something.large",
    resource={"prefect.resource.id": "test"},
    related=[{"prefect.resource.id": "test1", "prefect.resource.role": "test-role", "data": "#" * 100_000}],
)
```

But this will not:

```python
from prefect.events import emit_event

emit_event(
    event="something.large",
    resource={"prefect.resource.id": "test"},
    related=[{"prefect.resource.id": "test1", "prefect.resource.role": "test-role", "data": "#" * 1_000_000}],
)
```

I didn't know about `server.events.maximum_related_resources`, while it's probably sensible to have an upper limit here it's related to how the data actually gets stored in the index.

Please let me know if I can help in any way.

Commit message: Update maximum related resources to 100 for safety

Patch to review:
diff --git a/docs/v3/api-ref/settings-ref.mdx b/docs/v3/api-ref/settings-ref.mdx
index efdb415ecc00..1d770a1908ce 100644
--- a/docs/v3/api-ref/settings-ref.mdx
+++ b/docs/v3/api-ref/settings-ref.mdx
@@ -1393,7 +1393,7 @@ The maximum number of related resources an Event may have.
 
 **Type**: `integer`
 
-**Default**: `500`
+**Default**: `100`
 
 **TOML dotted key path**: `server.events.maximum_related_resources`
 
diff --git a/schemas/settings.schema.json b/schemas/settings.schema.json
index b5e62bbe3fc9..31b3e932a5e5 100644
--- a/schemas/settings.schema.json
+++ b/schemas/settings.schema.json
@@ -1254,7 +1254,7 @@
                     "type": "integer"
                 },
                 "maximum_related_resources": {
-                    "default": 500,
+                    "default": 100,
                     "description": "The maximum number of related resources an Event may have.",
                     "supported_environment_variables": [
                         "PREFECT_SERVER_EVENTS_MAXIMUM_RELATED_RESOURCES",
diff --git a/src/prefect/server/database/_migrations/MIGRATION-NOTES.md b/src/prefect/server/database/_migrations/MIGRATION-NOTES.md
index 999a8bb911ea..cd5d9434315c 100644
--- a/src/prefect/server/database/_migrations/MIGRATION-NOTES.md
+++ b/src/prefect/server/database/_migrations/MIGRATION-NOTES.md
@@ -8,6 +8,10 @@ Each time a database migration is written, an entry is included here with:
 
 This gives us a history of changes and will create merge conflicts if two migrations are made at once, flagging situations where a branch needs to be updated before merging.
 
+# Update `events` table `event_related_occurred` index for Postgres
+SQLite: None
+Postgres: `7a73514ca2d6`
+
 # Add `deployment_version` table
 SQLite: `bbca16f6f218`
 Postgres: `06b7c293bc69`
diff --git a/src/prefect/server/database/_migrations/versions/postgresql/2025_04_04_092158_7a73514ca2d6_add_ix_events__event_related_occurred_.py b/src/prefect/server/database/_migrations/versions/postgresql/2025_04_04_092158_7a73514ca2d6_add_ix_events__event_related_occurred_.py
new file mode 100644
index 000000000000..f7435891db18
--- /dev/null
+++ b/src/prefect/server/database/_migrations/versions/postgresql/2025_04_04_092158_7a73514ca2d6_add_ix_events__event_related_occurred_.py
@@ -0,0 +1,60 @@
+"""Add ix_events__event_related_occurred_md5
+
+Revision ID: 7a73514ca2d6
+Revises: 06b7c293bc69
+Create Date: 2025-04-04 09:21:58.070532
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "7a73514ca2d6"
+down_revision = "06b7c293bc69"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.drop_index("ix_events__event_related_occurred", table_name="events")
+    op.drop_index("ix_events__related_resource_ids", table_name="events")
+    op.create_index(
+        "ix_events__event_related_occurred_md5",
+        "events",
+        [sa.literal_column("md5(related::text)"), "event", "occurred"],
+        unique=False,
+        postgresql_using="btree",
+    )
+    op.create_index(
+        "ix_events__related_resource_ids_md5",
+        "events",
+        [sa.literal_column("md5(related_resource_ids::text)"), "event", "occurred"],
+        unique=False,
+        postgresql_using="btree",
+    )
+
+
+def downgrade():
+    op.drop_index(
+        "ix_events__event_related_occurred_md5",
+        table_name="events",
+        postgresql_using="btree",
+    )
+    op.drop_index(
+        "ix_events__related_resource_ids_md5",
+        table_name="events",
+        postgresql_using="btree",
+    )
+    op.create_index(
+        "ix_events__event_related_occurred",
+        "events",
+        ["event", "related", "occurred"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_events__related_resource_ids",
+        "events",
+        ["related_resource_ids"],
+        unique=False,
+    )
diff --git a/src/prefect/server/database/orm_models.py b/src/prefect/server/database/orm_models.py
index 0df613de9191..378ab914ed55 100644
--- a/src/prefect/server/database/orm_models.py
+++ b/src/prefect/server/database/orm_models.py
@@ -786,6 +786,23 @@ class DeploymentSchedule(Base):
     )
     slug: Mapped[Optional[str]] = mapped_column(sa.String, nullable=True)
 
+    @declared_attr.directive
+    @classmethod
+    def __table_args__(cls) -> Iterable[sa.Index]:
+        return (
+            sa.Index(
+                "ix_deployment_schedule__deployment_id__slug",
+                cls.deployment_id,
+                cls.slug,
+                unique=True,
+            ),
+            sa.Index(
+                "ix_deployment_schedule__slug",
+                cls.slug,
+                unique=False,
+            ),
+        )
+
 
 class Deployment(Base):
     """SQLAlchemy model of a deployment."""
@@ -1399,7 +1416,14 @@ def __tablename__(cls) -> str:
         return "events"
 
     __table_args__: Any = (
-        sa.Index("ix_events__related_resource_ids", "related_resource_ids"),
+        sa.Index("ix_events__related_resource_ids", "related_resource_ids").ddl_if(
+            dialect="sqlite"
+        ),
+        sa.Index(
+            "ix_events__related_resource_ids_md5",
+            sa.text("md5(related_resource_ids::text)"),
+            postgresql_using="btree",
+        ).ddl_if(dialect="postgresql"),
         sa.Index("ix_events__occurred", "occurred"),
         sa.Index("ix_events__event__id", "event", "id"),
         sa.Index(
@@ -1410,7 +1434,16 @@ def __tablename__(cls) -> str:
         ),
         sa.Index("ix_events__occurred_id", "occurred", "id"),
         sa.Index("ix_events__event_occurred_id", "event", "occurred", "id"),
-        sa.Index("ix_events__event_related_occurred", "event", "related", "occurred"),
+        sa.Index(
+            "ix_events__event_related_occurred", "event", "related", "occurred"
+        ).ddl_if(dialect="sqlite"),
+        sa.Index(
+            "ix_events__event_related_occurred_md5",
+            sa.text("md5(related::text)"),
+            "event",
+            "occurred",
+            postgresql_using="btree",
+        ).ddl_if(dialect="postgresql"),
     )
 
     occurred: Mapped[DateTime]
diff --git a/src/prefect/settings/models/server/events.py b/src/prefect/settings/models/server/events.py
index f50f10dd2f20..f3eeeb8cf12d 100644
--- a/src/prefect/settings/models/server/events.py
+++ b/src/prefect/settings/models/server/events.py
@@ -50,7 +50,7 @@ class ServerEventsSettings(PrefectBaseSettings):
     )
 
     maximum_related_resources: int = Field(
-        default=500,
+        default=100,
         description="The maximum number of related resources an Event may have.",
         validation_alias=AliasChoices(
             AliasPath("maximum_related_resources"),
diff --git a/tests/events/server/storage/test_event_persister.py b/tests/events/server/storage/test_event_persister.py
index fcf0ff17c643..09e28d2c1c29 100644
--- a/tests/events/server/storage/test_event_persister.py
+++ b/tests/events/server/storage/test_event_persister.py
@@ -1,4 +1,5 @@
 import asyncio
+import uuid
 from datetime import timedelta
 from typing import AsyncGenerator, Optional, Sequence
 from uuid import UUID, uuid4
@@ -12,7 +13,11 @@
 from prefect.server.database import PrefectDBInterface, db_injector
 from prefect.server.database.orm_models import ORMEventResource
 from prefect.server.events.filters import EventFilter
-from prefect.server.events.schemas.events import ReceivedEvent
+from prefect.server.events.schemas.events import (
+    ReceivedEvent,
+    RelatedResource,
+    Resource,
+)
 from prefect.server.events.services import event_persister
 from prefect.server.events.services.event_persister import batch_delete
 from prefect.server.events.storage.database import query_events, write_events
@@ -88,6 +93,31 @@ def event() -> ReceivedEvent:
     )
 
 
+@pytest.fixture
+def event_with_many_related_resources() -> ReceivedEvent:
+    return ReceivedEvent(
+        occurred=now("UTC"),
+        event="hello",
+        resource=Resource(
+            {"prefect.resource.id": "my.resource.id", "label-1": "value-1"}
+        ),
+        related=[
+            RelatedResource(
+                {
+                    "prefect.resource.id": str(uuid.uuid4()),
+                    "prefect.resource.role": "test.related",
+                    "data": "test.data",
+                }
+            )
+            for _ in range(99)
+        ],
+        payload={"hello": "world"},
+        received=DateTime(2022, 2, 3, 4, 5, 6, 7).astimezone(ZoneInfo("UTC")),
+        id=uuid4(),
+        follows=UUID("ffffffff-ffff-ffff-ffff-ffffffffffff"),
+    )
+
+
 @pytest.fixture
 def message(event: ReceivedEvent) -> Message:
     return CapturedMessage(
@@ -96,6 +126,16 @@ def message(event: ReceivedEvent) -> Message:
     )
 
 
+@pytest.fixture
+def message_with_many_related_resources(
+    event_with_many_related_resources: ReceivedEvent,
+) -> Message:
+    return CapturedMessage(
+        data=event_with_many_related_resources.model_dump_json().encode(),
+        attributes={},
+    )
+
+
 async def test_start_and_stop_service():
     service = event_persister.EventPersister()
     service_task = asyncio.create_task(service.start())
@@ -189,6 +229,24 @@ async def test_handling_message_writes_event_resources(
     assert related_3.resource == {"label-1": "value-5", "label-2": "value-6"}
 
 
+async def test_handling_message_writes_event_resources_with_many_related_resources(
+    frozen_time: DateTime,
+    db: PrefectDBInterface,
+    event_persister_handler: MessageHandler,
+    message_with_many_related_resources: Message,
+    session: AsyncSession,
+    event_with_many_related_resources: ReceivedEvent,
+):
+    await event_persister_handler(message_with_many_related_resources)
+
+    resources = await get_resources(session, event_with_many_related_resources.id, db)
+    assert len(resources) == 100
+
+    event = await get_event(event_with_many_related_resources.id)
+    assert event
+    assert event == event_with_many_related_resources
+
+
 @pytest.fixture
 def empty_message() -> Message:
     return CapturedMessage(


Code review comments:
diff_hunk: @@ -1399,7 +1416,14 @@ def __tablename__(cls) -> str:
         return "events"
 
     __table_args__: Any = (
-        sa.Index("ix_events__related_resource_ids", "related_resource_ids"),
+        sa.Index("ix_events__related_resource_ids", "related_resource_ids").ddl_if(
+            dialect="sqlite"
+        ),
+        sa.Index(
+            "ix_events__related_resource_ids_md5",
+            sa.text("md5(related_resource_ids::text)"),
+            postgresql_using="btree",
+        ).ddl_if(dialect="postgresql"),
path: src/prefect/server/database/orm_models.py
text: @user1:
When could this be used?  Unless we're updating the queries to use `where related_resource_ids = md5(...)`, then I don't think it will be in play?  And even then, I think an exact match on the resources IDs would be a pretty rare query.  Maybe just remove the index for now?

@author:
It gets used if you filter events by related resource ID. Postgres also suggests using a full-text index which might work better? 
```
2025-04-04 09:05:04.620 UTC [129] HINT:  Values larger than 1/3 of a buffer page cannot be indexed.
        Consider a function index of an MD5 hash of the value, or use full text indexing.
```

@user1:
I'm pretty sure that index as it stands can only be used in a query that said `where related_resource_ids = md5(...exactly the full list of ids in the same order...)`.

A GIN index does make sense there, I believe we had one in Prefect Cloud for a while when events were in PostgreSQL.

@author:
@user1 I updated the existing indexes to take advantage of GIN, which also seems to fix the original issue. Could you take a look at the new indexes?