ref(outcomes): add quantity64 in processor

rust_snuba/src/processors/outcomes.rs

@@ -82,6 +82,12 @@ pub fn process_message(
         }
     }
 
+    msg.quantity32 = Some(
+        msg.quantity
+            .and_then(|quantity| u32::try_from(quantity).ok())
+            .unwrap_or(0),
+    );
+
     InsertBatch::from_rows([msg], None)
 }
 
@@ -96,7 +102,11 @@ struct Outcome {
     timestamp: StringToIntDatetime,
     outcome: u8,
     category: Option<u8>,
-    quantity: Option<u32>,
+    #[serde(rename(serialize = "quantity64"))]
+    quantity: Option<u64>,
+    // Only derived from `quantity` in `process_message`
+    #[serde(skip_deserializing, rename(serialize = "quantity"))]
+    quantity32: Option<u32>,
     reason: Option<String>,
     event_id: Option<Uuid>,
 }
@@ -127,7 +137,32 @@ mod tests {
         let result = process_message(payload, meta, &ProcessorConfig::default())
             .expect("The message should be processed");
 
-        let expected = b"{\"org_id\":1,\"project_id\":1,\"key_id\":null,\"timestamp\":1680029444,\"outcome\":4,\"category\":1,\"quantity\":3,\"reason\":null,\"event_id\":null}\n";
+        let expected = b"{\"org_id\":1,\"project_id\":1,\"key_id\":null,\"timestamp\":1680029444,\"outcome\":4,\"category\":1,\"quantity64\":3,\"quantity\":3,\"reason\":null,\"event_id\":null}\n";
+
+        assert_eq!(result.rows.into_encoded_rows(), expected);
+    }
+
+    #[test]
+    fn test_outcome_quantity_overflow() {
+        // A quantity larger than u32::MAX cannot fit in the u32 `quantity`
+        // column; default to 0 while `quantity64` keeps the full value.
+        let data = r#"{
+            "org_id": 1,
+            "outcome": 4,
+            "project_id": 1,
+            "quantity": 5000000000,
+            "timestamp": "2023-03-28T18:50:44.000011Z"
+          }"#;
+        let payload = KafkaPayload::new(None, None, Some(data.as_bytes().to_vec()));
+        let meta = KafkaMessageMetadata {
+            partition: 0,
+            offset: 1,
+            timestamp: DateTime::from(SystemTime::now()),
+        };
+        let result = process_message(payload, meta, &ProcessorConfig::default())
+            .expect("The message should be processed");
+
+        let expected = b"{\"org_id\":1,\"project_id\":1,\"key_id\":null,\"timestamp\":1680029444,\"outcome\":4,\"category\":1,\"quantity64\":5000000000,\"quantity\":0,\"reason\":null,\"event_id\":null}\n";
 
         assert_eq!(result.rows.into_encoded_rows(), expected);
     }

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@outcomes-OutcomesProcessor-outcomes__1__outcomes-discarded-hash.json.snap

@@ -12,6 +12,7 @@ expression: snapshot_payload
     "outcome": 1,
     "project_id": 1,
     "quantity": 1,
+    "quantity64": 1,
     "reason": "discarded-hash",
     "timestamp": 1680029444
   }

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@outcomes-OutcomesProcessor-outcomes__1__outcomes-lb.json.snap

@@ -12,6 +12,7 @@ expression: snapshot_payload
     "outcome": 4,
     "project_id": 1,
     "quantity": 1,
+    "quantity64": 1,
     "reason": null,
     "timestamp": 1680029439
   }

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@outcomes-OutcomesProcessor-outcomes__1__outcomes-null-values.json.snap

@@ -12,6 +12,7 @@ expression: snapshot_payload
     "outcome": 0,
     "project_id": 1,
     "quantity": 1,
+    "quantity64": 1,
     "reason": null,
     "timestamp": 1679686100
   }

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@outcomes-OutcomesProcessor-outcomes__1__outcomes-pop-us.json.snap

@@ -12,6 +12,7 @@ expression: snapshot_payload
     "outcome": 3,
     "project_id": 1,
     "quantity": 1,
+    "quantity64": 1,
     "reason": "project_id",
     "timestamp": 1680043860
   }

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@outcomes-OutcomesProcessor-outcomes__1__outcomes-relay-internal.json.snap

@@ -12,6 +12,7 @@ expression: snapshot_payload
     "outcome": 3,
     "project_id": 1,
     "quantity": 1,
+    "quantity64": 1,
     "reason": "project_id",
     "timestamp": 1680043980
   }

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@outcomes-OutcomesProcessor-outcomes__1__outcomes2-missing-key-id.json.snap

@@ -12,6 +12,7 @@ expression: snapshot_payload
     "outcome": 4,
     "project_id": 1,
     "quantity": 3,
+    "quantity64": 3,
     "reason": null,
     "timestamp": 1680029449
   }