Checkpoint.

lib/cdc/util/relational_event.go

@@ -82,14 +82,16 @@ func (s *SchemaEventPayload) GetTableName() string {
 func (s *SchemaEventPayload) GetData(pkMap map[string]any, tc *kafkalib.TopicConfig) map[string]any {
 	var retMap map[string]any
 	if len(s.Payload.After) == 0 {
+		if len(s.Payload.Before) > 0 {
+			retMap = s.Payload.Before
+		} else {
+			retMap = make(map[string]any)
+		}
 		// This is a delete payload, so mark it as deleted.
 		// And we need to reconstruct the data bit since it will be empty.
 		// We _can_ rely on *before* since even without running replicate identity, it will still copy over
 		// the PK. We can explore simplifying this interface in the future by leveraging before.
-		retMap = map[string]any{
-			constants.DeleteColumnMarker: true,
-		}
-
+		retMap[constants.DeleteColumnMarker] = true
 		for k, v := range pkMap {
 			retMap[k] = v
 		}