diff --git a/snuba/migrations/group_loader.py b/snuba/migrations/group_loader.py
index faa4c5beea..2a99fa2206 100644
--- a/snuba/migrations/group_loader.py
+++ b/snuba/migrations/group_loader.py
@@ -378,6 +378,7 @@ def get_migrations(self) -> Sequence[str]:
             "0010_spans_add_compression",
             "0011_spans_add_index_on_trace_id",
             "0012_spans_add_index_on_transaction_name",
+            "0013_spans_add_indexes_for_tag_columns",
         ]
 
 
diff --git a/snuba/snuba_migrations/spans/0013_spans_add_indexes_for_tag_columns.py b/snuba/snuba_migrations/spans/0013_spans_add_indexes_for_tag_columns.py
new file mode 100644
index 0000000000..3bc309b058
--- /dev/null
+++ b/snuba/snuba_migrations/spans/0013_spans_add_indexes_for_tag_columns.py
@@ -0,0 +1,47 @@
+from typing import List, Sequence
+
+from snuba.clusters.storage_sets import StorageSetKey
+from snuba.migrations import migration, operations
+
+table_name_prefix = "spans"
+storage_set = StorageSetKey.SPANS
+indexes = [
+    ("bf_tags_key", "tags.key", "bloom_filter(0.0)", 1),
+    ("bf_tags_hash_map", "_tags_hash_map", "bloom_filter(0.0)", 1),
+    ("bf_sentry_tags_hash_map", "_sentry_tags_hash_map", "bloom_filter(0.0)", 1),
+]
+
+
+class Migration(migration.ClickhouseNodeMigration):
+    blocking = False
+
+    def forwards_ops(self) -> Sequence[operations.SqlOperation]:
+        ops: List[operations.SqlOperation] = []
+        for index in indexes:
+            index_name, index_expression, index_type, granularity = index
+            ops.append(
+                operations.AddIndex(
+                    storage_set=storage_set,
+                    table_name=f"{table_name_prefix}_local",
+                    index_name=index_name,
+                    index_expression=index_expression,
+                    index_type=index_type,
+                    granularity=granularity,
+                    target=operations.OperationTarget.LOCAL,
+                )
+            )
+        return ops
+
+    def backwards_ops(self) -> Sequence[operations.SqlOperation]:
+        ops: List[operations.SqlOperation] = []
+        for index in indexes:
+            index_name, _, _, _ = index
+            ops.append(
+                operations.DropIndex(
+                    storage_set=storage_set,
+                    table_name=f"{table_name_prefix}_local",
+                    index_name=index_name,
+                    target=operations.OperationTarget.LOCAL,
+                )
+            )
+        return ops