perf: Optimize content_type filtering with computed column and indexes (#868)

Author: aliel

deployment/migrations/versions/0040_83a04f64a1db_add_content_type_as_field.py

@@ -0,0 +1,34 @@
+"""Add content>>type as field
+
+Revision ID: 83a04f64a1db
+Revises: d0e1f2a3b4c5
+Create Date: 2025-10-14 09:26:24.239634
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '83a04f64a1db'
+down_revision = 'd0e1f2a3b4c5'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column('messages', sa.Column('content_type', sa.String(), sa.Computed("content->>'type'", persisted=True), nullable=True))
+    op.create_index('ix_messages_content_type', 'messages', ['content_type'], unique=False)
+    # Add an index on sender + content.type as content.type are often used with the sender together
+    op.create_index(
+        'ix_messages_sender_content_type',
+        'messages',
+        ['sender', 'content_type'],
+        unique=False
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('ix_messages_sender_content_type', 'messages')
+    op.drop_index('ix_messages_content_type', 'messages')
+    op.drop_column('messages', 'content_type')

src/aleph/db/accessors/messages.py

@@ -130,9 +130,7 @@ def make_matching_messages_query(
             MessageDb.content["item_hash"].astext.in_(content_hashes)
         )
     if content_types:
-        select_stmt = select_stmt.where(
-            MessageDb.content["type"].astext.in_(content_types)
-        )
+        select_stmt = select_stmt.where(MessageDb.content_type.in_(content_types))
     if tags:
         select_stmt = select_stmt.where(
             MessageDb.content["content"]["tags"].has_any(array(tags))
@@ -316,7 +314,7 @@ def get_unconfirmed_messages(
 def make_message_upsert_query(message: MessageDb) -> Insert:
     return (
         insert(MessageDb)
-        .values(message.to_dict())
+        .values(message.to_dict(exclude={"content_type"}))
         .on_conflict_do_update(
             constraint="messages_pkey",
             set_={"time": func.least(MessageDb.time, message.time)},

src/aleph/db/models/messages.py

@@ -18,7 +18,9 @@
     ARRAY,
     TIMESTAMP,
     Column,
+    Computed,
     ForeignKey,
+    Index,
     Integer,
     String,
     Table,
@@ -99,6 +101,14 @@ class MessageDb(Base):
     time: dt.datetime = Column(TIMESTAMP(timezone=True), nullable=False, index=True)
     channel: Optional[Channel] = Column(String, nullable=True, index=True)
     size: int = Column(Integer, nullable=False)
+    content_type: Optional[str] = Column(
+        String, Computed("content->>'type'", persisted=True)
+    )
+
+    __table_args__ = (
+        Index("ix_messages_content_type", content_type),
+        Index("ix_messages_sender_content_type", sender, content_type),
+    )
 
     confirmations: "List[ChainTxDb]" = relationship(
         "ChainTxDb", secondary=message_confirmations