Run migration for exports by requestor column (#29515)

Filter out non snapshotExport sources by default once we run the
migration

GitOrigin-RevId: 4ec2243fad60bdace69d8da68f35e6d6e5e1b498

Author: nipunn1313

crates/model/src/exports/mod.rs

@@ -40,25 +40,37 @@ pub static EXPORTS_TABLE: LazyLock<TableName> =
 pub static EXPORTS_BY_STATE_AND_TS_INDEX: LazyLock<IndexName> =
     LazyLock::new(|| system_index(&EXPORTS_TABLE, "by_state_and_ts"));
 
+pub static EXPORTS_BY_REQUESTOR: LazyLock<IndexName> =
+    LazyLock::new(|| system_index(&EXPORTS_TABLE, "by_requestor"));
+
 pub static EXPORTS_STATE_FIELD: LazyLock<FieldPath> =
     LazyLock::new(|| "state".parse().expect("Invalid built-in field"));
 
 pub static EXPORTS_TS_FIELD: LazyLock<FieldPath> =
     LazyLock::new(|| "start_ts".parse().expect("Invalid built-in field"));
 
+static EXPORTS_REQUESTOR_FIELD: LazyLock<FieldPath> =
+    LazyLock::new(|| "requestor".parse().expect("Invalid built-in field"));
+
 pub struct ExportsTable;
 impl SystemTable for ExportsTable {
     fn table_name(&self) -> &'static TableName {
         &EXPORTS_TABLE
     }
 
     fn indexes(&self) -> Vec<SystemIndex> {
-        vec![SystemIndex {
-            name: EXPORTS_BY_STATE_AND_TS_INDEX.clone(),
-            fields: vec![EXPORTS_STATE_FIELD.clone(), EXPORTS_TS_FIELD.clone()]
-                .try_into()
-                .unwrap(),
-        }]
+        vec![
+            SystemIndex {
+                name: EXPORTS_BY_STATE_AND_TS_INDEX.clone(),
+                fields: vec![EXPORTS_STATE_FIELD.clone(), EXPORTS_TS_FIELD.clone()]
+                    .try_into()
+                    .unwrap(),
+            },
+            SystemIndex {
+                name: EXPORTS_BY_REQUESTOR.clone(),
+                fields: vec![EXPORTS_REQUESTOR_FIELD.clone()].try_into().unwrap(),
+            },
+        ]
     }
 
     fn validate_document(&self, document: ResolvedDocument) -> anyhow::Result<()> {