diff --git a/snuba/datasets/configuration/issues/entities/search_issues.yaml b/snuba/datasets/configuration/issues/entities/search_issues.yaml
index 9b31161336..e82678c3de 100644
--- a/snuba/datasets/configuration/issues/entities/search_issues.yaml
+++ b/snuba/datasets/configuration/issues/entities/search_issues.yaml
@@ -36,6 +36,24 @@ schema:
             [ { name: key, type: String }, { name: value, type: String } ],
         },
     },
+    {
+      name: flags,
+      type: Nested,
+      args:
+        {
+          subcolumns:
+            [ { name: key, type: String }, { name: value, type: String } ],
+        },
+    },
+    {
+      name: _flags_hash_map,
+      type: Array,
+      args:
+        {
+          schema_modifiers: [ readonly ],
+          inner_type: { type: UInt, args: { size: 64 } },
+        },
+    },
     { name: user, type: String, args: { schema_modifiers: [ nullable ] } },
     { name: user_hash, type: UInt, args: { size: 64, schema_modifiers: [ nullable ] } },
     { name: user_id, type: String, args: { schema_modifiers: [ nullable ] } },
@@ -173,6 +191,18 @@ storages:
             from_col_name: tags_value
             to_function_name: arrayJoin
             to_function_column: tags.value
+        - mapper: ColumnToFunctionOnColumn
+          args:
+            from_table_name: null
+            from_col_name: flags_key
+            to_function_name: arrayJoin
+            to_function_column: flags.key
+        - mapper: ColumnToFunctionOnColumn
+          args:
+            from_table_name: null
+            from_col_name: flags_value
+            to_function_name: arrayJoin
+            to_function_column: flags.value
       subscriptables:
         - mapper: SubscriptableMapper
           args:
@@ -186,6 +216,12 @@ storages:
             from_column_name: contexts
             to_nested_col_table:
             to_nested_col_name: contexts
+        - mapper: SubscriptableMapper
+          args:
+            from_column_table:
+            from_column_name: flags
+            to_nested_col_table:
+            to_nested_col_name: flags
 
 storage_selector:
   selector: DefaultQueryStorageSelector
diff --git a/snuba/datasets/configuration/issues/storages/search_issues.yaml b/snuba/datasets/configuration/issues/storages/search_issues.yaml
index 8c94760a23..d8777d28d0 100644
--- a/snuba/datasets/configuration/issues/storages/search_issues.yaml
+++ b/snuba/datasets/configuration/issues/storages/search_issues.yaml
@@ -38,6 +38,9 @@ schema:
       { name: tags.key, type: Array, args: { inner_type: { type: String } } },
       { name: tags.value, type: Array, args: { inner_type: { type: String } } },
       { name: _tags_hash_map, type: Array, args: { inner_type: { type: UInt, args: { size: 64 } } } },
+      { name: flags.key, type: Array, args: { inner_type: { type: String } } },
+      { name: flags.value, type: Array, args: { inner_type: { type: String } } },
+      { name: _flags_hash_map, type: Array, args: { inner_type: { type: UInt, args: { size: 64 } } } },
       { name: user, type: String, args: { schema_modifiers: [ nullable ] } },
       { name: user_hash, type: UInt, args: { size: 64, schema_modifiers: [ nullable ] } },
       { name: user_id, type: String, args: { schema_modifiers: [ nullable ] } },
@@ -126,6 +129,17 @@ query_processors:
   - processor: ArrayJoinKeyValueOptimizer
     args:
       column_name: tags
+  - processor: MappingOptimizer
+    args:
+      column_name: flags
+      hash_map_name: _flags_hash_map
+      killswitch: search_issues_flags_hash_map_enabled
+  - processor: EmptyTagConditionProcessor
+    args:
+      column_name: flags.key
+  - processor: ArrayJoinKeyValueOptimizer
+    args:
+      column_name: flags
   - processor: UUIDColumnProcessor
     args:
       columns: [occurrence_id, event_id, trace_id, profile_id, replay_id]
diff --git a/snuba/datasets/processors/search_issues_processor.py b/snuba/datasets/processors/search_issues_processor.py
index e3e76fef6d..f822ae5ba3 100644
--- a/snuba/datasets/processors/search_issues_processor.py
+++ b/snuba/datasets/processors/search_issues_processor.py
@@ -21,6 +21,7 @@
     extract_extra_contexts,
     extract_extra_tags,
     extract_http,
+    extract_nested,
     extract_user,
 )
 from snuba.datasets.processors import DatasetMessageProcessor
@@ -66,6 +67,7 @@ class IssueEventData(TypedDict, total=False):
     start_timestamp: float
 
     tags: Mapping[str, Any]
+    flags: Mapping[str, Any]
     user: Mapping[str, Any]  # user_hash, user_id, user_name, user_email, ip_address
     sdk: Mapping[str, Any]  # sdk_name, sdk_version
     contexts: Mapping[str, Any]
@@ -157,6 +159,18 @@ def _process_tags(
             promoted_tags.get("sentry:dist", event_data.get("dist")),
         )
 
+    def _process_flags(
+        self, event_data: IssueEventData, processed: MutableMapping[str, Any]
+    ) -> None:
+        existing_flags = event_data.get("flags", None)
+        flags: Mapping[str, Any] = _as_dict_safe(cast(Dict[str, Any], existing_flags))
+        if not existing_flags:
+            processed["flags.key"], processed["flags.value"] = [], []
+        else:
+            processed["flags.key"], processed["flags.value"] = extract_nested(
+                flags, lambda s: _unicodify(s) or None
+            )
+
     def _process_request_data(
         self, event_data: IssueEventData, processed: MutableMapping[str, Any]
     ) -> None:
@@ -291,6 +305,7 @@ def process_insert_v1(
         self._process_tags(
             event_data, fields
         )  # environment, release, dist, user, tags.key, tags.value
+        self._process_flags(event_data, fields)  # flags.key, flags.value
         self._process_user(
             event_data, fields
         )  # user_name, user_id, user_email, ip_address_v4/ip_address_v6