| 88 | |
| 89 | #[test] |
| 90 | fn log_compaction_keeps_latest_per_key() { |
| 91 | let config = CompactionConfig::key("id"); |
| 92 | let buf = StreamBuffer::new( |
| 93 | "users_stream".to_string(), |
| 94 | RetentionConfig { |
| 95 | max_events: 1_000_000, |
| 96 | max_age_secs: 86_400, |
| 97 | }, |
| 98 | ); |
| 99 | |
| 100 | // Two events for same row_id but different sequences. |
| 101 | buf.push(CdcEvent { |
| 102 | sequence: 1, |
| 103 | partition: 0, |
| 104 | collection: "users".into(), |
| 105 | op: "INSERT".into(), |
| 106 | row_id: "u-1".into(), |
| 107 | event_time: now_ms(), |
| 108 | lsn: 10, |
| 109 | tenant_id: 1, |
| 110 | new_value: Some(serde_json::json!({"id": "u-1", "name": "Alice"})), |
| 111 | old_value: None, |
| 112 | schema_version: 0, |
| 113 | field_diffs: None, |
| 114 | system_time_ms: None, |
| 115 | valid_time_ms: None, |
| 116 | }); |
| 117 | buf.push(CdcEvent { |
| 118 | sequence: 2, |
| 119 | partition: 0, |
| 120 | collection: "users".into(), |
| 121 | op: "UPDATE".into(), |
| 122 | row_id: "u-1".into(), |
| 123 | event_time: now_ms(), |
| 124 | lsn: 20, |
| 125 | tenant_id: 1, |
| 126 | new_value: Some(serde_json::json!({"id": "u-1", "name": "Bob"})), |
| 127 | old_value: None, |
| 128 | schema_version: 0, |
| 129 | field_diffs: None, |
| 130 | system_time_ms: None, |
| 131 | valid_time_ms: None, |
| 132 | }); |
| 133 | |
| 134 | // Before compaction: both events present. |
| 135 | assert_eq!(buf.read_from_lsn(0, 100).len(), 2); |
| 136 | |
| 137 | // Compact. |
| 138 | buf.compact(&config.key_field, config.tombstone_grace_secs); |
| 139 | |
| 140 | // After compaction: only latest event per row_id. |
| 141 | let events = buf.read_from_lsn(0, 100); |
| 142 | assert_eq!(events.len(), 1); |
| 143 | assert_eq!(events[0].sequence, 2); |
| 144 | assert_eq!(events[0].op, "UPDATE"); |
| 145 | } |
| 146 | |
| 147 | #[test] |