1use mz_expr_derive::sqlfunc;
30use mz_repr::adt::jsonb::{Jsonb, JsonbRef};
31use mz_repr::{Datum, Row, RowPacker};
32
33use crate::EvalError;
34
35#[sqlfunc]
44fn parse_catalog_audit_log_details<'a>(a: JsonbRef<'a>) -> Result<Jsonb, EvalError> {
45 const FLATTENED_FIELDS: &[(&str, &str, &str, Option<&str>)] = &[
49 ("IdFullNameV1", "", "name", None),
50 ("CreateSourceSinkV1", "", "name", None),
51 ("CreateSourceSinkV2", "", "name", None),
52 ("CreateSourceSinkV3", "", "name", None),
53 ("CreateSourceSinkV4", "", "name", None),
54 ("CreateIndexV1", "", "name", None),
55 ("CreateMaterializedViewV1", "", "name", None),
56 ("AlterSourceSinkV1", "", "name", None),
57 ("AlterSetClusterV1", "", "name", None),
58 ("UpdateItemV1", "", "name", None),
59 (
62 "AlterApplyReplacementV1",
63 "",
64 "target",
65 Some("IdFullNameV1"),
66 ),
67 ];
68
69 const RENAMES: &[(&str, &str, &str, &str)] = &[
72 ("CreateSourceSinkV2", "", "external_type", "type"),
73 ("CreateSourceSinkV3", "", "external_type", "type"),
74 ("CreateSourceSinkV4", "", "external_type", "type"),
75 (
76 "RefreshDecisionWithReasonV1",
77 "",
78 "rehydration_time_estimate",
79 "hydration_time_estimate",
80 ),
81 (
82 "RefreshDecisionWithReasonV2",
83 "",
84 "rehydration_time_estimate",
85 "hydration_time_estimate",
86 ),
87 ];
88
89 const DROP_NULL: &[(&str, &str, &str)] = &[
93 ("CreateClusterReplicaV1", "", "replica_id"),
94 ("DropClusterReplicaV1", "", "replica_id"),
95 ("CreateClusterReplicaV2", "", "scheduling_policies"),
96 ("CreateClusterReplicaV3", "", "scheduling_policies"),
97 ("CreateClusterReplicaV4", "", "scheduling_policies"),
98 ("DropClusterReplicaV2", "", "scheduling_policies"),
99 ("DropClusterReplicaV3", "", "scheduling_policies"),
100 ("CreateMaterializedViewV1", "", "replacement_target_id"),
101 ];
102
103 const REASON_MAP: &[(&str, &str)] = &[
106 ("Manual", "manual"),
107 ("Schedule", "schedule"),
108 ("System", "system"),
109 ("Reconfiguration", "reconfiguration"),
110 ("HydrationBurst", "hydration-burst"),
111 ("Retired", "retired"),
112 ];
113
114 #[allow(dead_code)]
120 #[derive(Copy, Clone)]
121 enum WrapKind {
122 Single,
123 Double,
124 }
125 const ENUM_COLLAPSE: &[(&str, &str, &str, WrapKind, &[(&str, &str)])] = &[
126 (
127 "RefreshDecisionWithReasonV1",
128 "",
129 "decision",
130 WrapKind::Single,
131 &[("On", "on"), ("Off", "off")],
132 ),
133 (
134 "RefreshDecisionWithReasonV2",
135 "",
136 "decision",
137 WrapKind::Single,
138 &[("On", "on"), ("Off", "off")],
139 ),
140 (
141 "CreateClusterReplicaV2",
142 "",
143 "reason",
144 WrapKind::Double,
145 REASON_MAP,
146 ),
147 (
148 "CreateClusterReplicaV3",
149 "",
150 "reason",
151 WrapKind::Double,
152 REASON_MAP,
153 ),
154 (
155 "CreateClusterReplicaV4",
156 "",
157 "reason",
158 WrapKind::Double,
159 REASON_MAP,
160 ),
161 (
162 "DropClusterReplicaV2",
163 "",
164 "reason",
165 WrapKind::Double,
166 REASON_MAP,
167 ),
168 (
169 "DropClusterReplicaV3",
170 "",
171 "reason",
172 WrapKind::Double,
173 REASON_MAP,
174 ),
175 (
176 "AlterClusterReconfigurationV1",
177 "",
178 "transition",
179 WrapKind::Double,
180 &[
181 ("Started", "started"),
182 ("Finalized", "finalized"),
183 ("TimedOut", "timed-out"),
184 ("Cancelled", "cancelled"),
185 ],
186 ),
187 (
188 "ClusterHydrationBurstV1",
189 "",
190 "transition",
191 WrapKind::Double,
192 &[("Started", "started"), ("Finished", "finished")],
193 ),
194 ];
195
196 const DESCEND: &[(&str, &str, &str, &str)] = &[
202 ("AlterApplyReplacementV1", "", "replacement", "IdFullNameV1"),
203 (
204 "CreateClusterReplicaV2",
205 "",
206 "scheduling_policies",
207 "SchedulingDecisionsWithReasonsV1",
208 ),
209 (
210 "CreateClusterReplicaV3",
211 "",
212 "scheduling_policies",
213 "SchedulingDecisionsWithReasonsV2",
214 ),
215 (
216 "CreateClusterReplicaV4",
217 "",
218 "scheduling_policies",
219 "SchedulingDecisionsWithReasonsV2",
220 ),
221 (
222 "DropClusterReplicaV2",
223 "",
224 "scheduling_policies",
225 "SchedulingDecisionsWithReasonsV1",
226 ),
227 (
228 "DropClusterReplicaV3",
229 "",
230 "scheduling_policies",
231 "SchedulingDecisionsWithReasonsV2",
232 ),
233 (
234 "SchedulingDecisionsWithReasonsV1",
235 "",
236 "on_refresh",
237 "RefreshDecisionWithReasonV1",
238 ),
239 (
240 "SchedulingDecisionsWithReasonsV2",
241 "",
242 "on_refresh",
243 "RefreshDecisionWithReasonV2",
244 ),
245 ];
246
247 fn lookup_flatten(variant: &str, path: &str, field: &str) -> Option<Option<&'static str>> {
248 FLATTENED_FIELDS
249 .iter()
250 .find(|(v, p, f, _)| *v == variant && *p == path && *f == field)
251 .map(|(_, _, _, sv)| *sv)
252 }
253
254 fn lookup_rename(variant: &str, path: &str, field: &str) -> Option<&'static str> {
255 RENAMES
256 .iter()
257 .find(|(v, p, k, _)| *v == variant && *p == path && *k == field)
258 .map(|(_, _, _, out)| *out)
259 }
260
261 fn is_drop_null(variant: &str, path: &str, field: &str) -> bool {
262 DROP_NULL
263 .iter()
264 .any(|(v, p, f)| *v == variant && *p == path && *f == field)
265 }
266
267 fn lookup_enum_collapse(
268 variant: &str,
269 path: &str,
270 field: &str,
271 ) -> Option<(WrapKind, &'static [(&'static str, &'static str)])> {
272 ENUM_COLLAPSE
273 .iter()
274 .find(|(v, p, f, _, _)| *v == variant && *p == path && *f == field)
275 .map(|(_, _, _, wrap, map)| (*wrap, *map))
276 }
277
278 fn lookup_descend(variant: &str, path: &str, field: &str) -> Option<&'static str> {
279 DESCEND
280 .iter()
281 .find(|(v, p, f, _)| *v == variant && *p == path && *f == field)
282 .map(|(_, _, _, sv)| *sv)
283 }
284
285 fn collapse_enum(
289 datum: Datum,
290 wrap: WrapKind,
291 map: &[(&'static str, &'static str)],
292 context: &str,
293 ) -> Result<&'static str, String> {
294 let inner = match wrap {
295 WrapKind::Single => datum,
296 WrapKind::Double => {
297 let Datum::Map(outer) = datum else {
298 return Err(format!(
299 "{context}: double-wrap enum expected outer object, got {datum:?}"
300 ));
301 };
302 let mut iter = outer.iter();
303 let (Some((_, v)), None) = (iter.next(), iter.next()) else {
304 return Err(format!(
305 "{context}: double-wrap enum expected single-key outer object"
306 ));
307 };
308 v
309 }
310 };
311 let Datum::Map(dict) = inner else {
312 return Err(format!(
313 "{context}: enum collapse expected object, got {inner:?}"
314 ));
315 };
316 let mut iter = dict.iter();
317 let (Some((variant_key, payload)), None) = (iter.next(), iter.next()) else {
318 return Err(format!(
319 "{context}: enum collapse expected single-key object"
320 ));
321 };
322 if let Datum::Map(payload_dict) = payload {
325 if payload_dict.iter().next().is_some() {
326 return Err(format!(
327 "{context}: enum variant {variant_key} carries non-empty payload"
328 ));
329 }
330 } else {
331 return Err(format!(
332 "{context}: enum variant {variant_key} payload is not an object"
333 ));
334 }
335 map.iter()
336 .find(|(k, _)| *k == variant_key)
337 .map(|(_, kebab)| *kebab)
338 .ok_or_else(|| format!("{context}: unknown enum variant {variant_key}"))
339 }
340
341 fn extend_path(path: &str, field: &str) -> String {
342 if path.is_empty() {
343 field.to_string()
344 } else {
345 format!("{path}.{field}")
346 }
347 }
348
349 fn rewrite(datum: Datum, variant: &str, path: &str, out: &mut RowPacker) -> Result<(), String> {
351 match datum {
352 Datum::Map(dict) => {
353 let mut iter = dict.iter();
355 if let (Some((k, v)), None) = (iter.next(), iter.next()) {
356 if k == "inner" {
357 return rewrite(v, variant, path, out);
358 }
359 }
360 rewrite_map(dict, variant, path, out)
361 }
362 Datum::List(list) => {
363 let mut result: Result<(), String> = Ok(());
364 out.push_list_with(|out| {
365 for v in list.iter() {
366 if let Err(e) = rewrite(v, variant, path, out) {
367 result = Err(e);
368 break;
369 }
370 }
371 });
372 result
373 }
374 other => {
375 out.push(other);
376 Ok(())
377 }
378 }
379 }
380
381 fn rewrite_map(
382 dict: mz_repr::DatumMap,
383 variant: &str,
384 path: &str,
385 out: &mut RowPacker,
386 ) -> Result<(), String> {
387 let mut entries = collect_entries(dict, variant, path)?;
388 entries.sort_by(|a, b| a.0.cmp(&b.0));
390 out.push_dict_with(|out| {
391 for (k, temp) in &entries {
392 out.push(Datum::String(k));
393 out.push(temp.unpack_first());
394 }
395 });
396 Ok(())
397 }
398
399 fn collect_entries(
403 dict: mz_repr::DatumMap,
404 variant: &str,
405 path: &str,
406 ) -> Result<Vec<(String, Row)>, String> {
407 let mut entries: Vec<(String, Row)> = Vec::new();
408
409 for (k, v) in dict.iter() {
410 if is_drop_null(variant, path, k) && matches!(v, Datum::JsonNull) {
411 continue;
412 }
413
414 if let Some(sub_variant_opt) = lookup_flatten(variant, path, k) {
416 let sub_variant = sub_variant_opt.unwrap_or("");
417 let Datum::Map(sub) = v else {
418 return Err(format!(
419 "expected flatten target {variant}.{k} to be an object"
420 ));
421 };
422 entries.extend(collect_entries(sub, sub_variant, "")?);
423 continue;
424 }
425
426 let out_key = lookup_rename(variant, path, k).unwrap_or(k).to_string();
427
428 if let Some((wrap, map)) = lookup_enum_collapse(variant, path, k) {
430 let kebab = collapse_enum(v, wrap, map, &format!("{variant}.{k}"))?;
431 let mut temp = Row::default();
432 temp.packer().push(Datum::String(kebab));
433 entries.push((out_key, temp));
434 continue;
435 }
436
437 if let Some(sub_variant) = lookup_descend(variant, path, k) {
440 let mut temp = Row::default();
441 rewrite(v, sub_variant, "", &mut temp.packer())?;
442 entries.push((out_key, temp));
443 continue;
444 }
445
446 let child_path = extend_path(path, k);
448 let mut temp = Row::default();
449 rewrite(v, variant, &child_path, &mut temp.packer())?;
450 entries.push((out_key, temp));
451 }
452
453 Ok(entries)
454 }
455
456 let parse = || -> Result<Jsonb, String> {
457 let Datum::Map(dict) = a.into_datum() else {
458 return Err("expected object".into());
459 };
460 let mut iter = dict.iter();
461 let (variant, inner) = iter
462 .next()
463 .ok_or_else(|| "empty details enum".to_string())?;
464 if iter.next().is_some() {
465 return Err("details enum had multiple keys".into());
466 }
467 let mut row = Row::default();
468 if variant == "ResetAllV1" {
471 row.packer().push(Datum::JsonNull);
472 return Ok(Jsonb::from_row(row));
473 }
474 let Datum::Map(_) = inner else {
475 return Err(format!("expected inner object for variant {variant}"));
476 };
477 rewrite(inner, variant, "", &mut row.packer())?;
478 Ok(Jsonb::from_row(row))
479 };
480
481 parse().map_err(|e| EvalError::InvalidCatalogJson(e.into()))
482}
483
484#[cfg(test)]
485mod tests {
486 use super::*;
487
488 fn check(input: &str, expected: &str) {
491 let input: Jsonb = input.parse().expect("valid input JSONB");
492 let actual = parse_catalog_audit_log_details(input.as_ref())
493 .expect("helper succeeded")
494 .to_string();
495 let actual_value: serde_json::Value =
496 serde_json::from_str(&actual).expect("valid output JSON");
497 let expected_value: serde_json::Value =
498 serde_json::from_str(expected).expect("valid expected JSON");
499 assert_eq!(actual_value, expected_value);
500 }
501
502 fn check_err(input: &str, expected_substr: &str) {
505 let input: Jsonb = input.parse().expect("valid input JSONB");
506 let err = parse_catalog_audit_log_details(input.as_ref()).expect_err("helper should error");
507 let msg = format!("{err:?}");
508 assert!(
509 msg.contains(expected_substr),
510 "error did not contain {expected_substr:?}: {msg}",
511 );
512 }
513
514 #[mz_ore::test]
517 fn variant_strip() {
518 check(
519 r#"{"IdNameV1": {"id": "u1", "name": "foo"}}"#,
520 r#"{"id": "u1", "name": "foo"}"#,
521 );
522 }
523
524 #[mz_ore::test]
527 fn flatten_name() {
528 check(
529 r#"{"IdFullNameV1": {
530 "id": "u1",
531 "name": {"database": "materialize", "schema": "public", "item": "t"}
532 }}"#,
533 r#"{"id": "u1", "database": "materialize", "schema": "public", "item": "t"}"#,
534 );
535 }
536
537 #[mz_ore::test]
542 fn flatten_recursive_and_nested_full_name() {
543 check(
544 r#"{"AlterApplyReplacementV1": {
545 "target": {
546 "id": "u1",
547 "name": {"database": "materialize", "schema": "public", "item": "mv"}
548 },
549 "replacement": {
550 "id": "u2",
551 "name": {"database": "materialize", "schema": "public", "item": "rp"}
552 }
553 }}"#,
554 r#"{
555 "id": "u1",
556 "database": "materialize",
557 "schema": "public",
558 "item": "mv",
559 "replacement": {
560 "id": "u2",
561 "database": "materialize",
562 "schema": "public",
563 "item": "rp"
564 }
565 }"#,
566 );
567 }
568
569 #[mz_ore::test]
573 fn string_wrapper_unwrap() {
574 check(
575 r#"{"AlterDefaultPrivilegeV1": {
576 "role_id": "u1",
577 "database_id": {"inner": "u2"},
578 "schema_id": {"inner": "u3"},
579 "grantee_id": "p",
580 "privileges": "r"
581 }}"#,
582 r#"{
583 "role_id": "u1",
584 "database_id": "u2",
585 "schema_id": "u3",
586 "grantee_id": "p",
587 "privileges": "r"
588 }"#,
589 );
590 }
591
592 #[mz_ore::test]
594 fn null_option() {
595 check(
596 r#"{"AlterDefaultPrivilegeV1": {
597 "role_id": "u1",
598 "database_id": null,
599 "schema_id": null,
600 "grantee_id": "p",
601 "privileges": "r"
602 }}"#,
603 r#"{
604 "role_id": "u1",
605 "database_id": null,
606 "schema_id": null,
607 "grantee_id": "p",
608 "privileges": "r"
609 }"#,
610 );
611 }
612
613 #[mz_ore::test]
619 fn nested_full_name_stays_nested() {
620 check(
621 r#"{"RenameItemV1": {
622 "id": "u1",
623 "old_name": {"database": "d", "schema": "s", "item": "a"},
624 "new_name": {"database": "d", "schema": "s", "item": "b"}
625 }}"#,
626 r#"{
627 "id": "u1",
628 "old_name": {"database": "d", "schema": "s", "item": "a"},
629 "new_name": {"database": "d", "schema": "s", "item": "b"}
630 }"#,
631 );
632 }
633
634 #[mz_ore::test]
637 fn rename_external_type() {
638 check(
639 r#"{"CreateSourceSinkV2": {
640 "id": "u1",
641 "name": {"database": "d", "schema": "s", "item": "src"},
642 "size": {"inner": "small"},
643 "external_type": "kafka"
644 }}"#,
645 r#"{
646 "id": "u1",
647 "database": "d",
648 "schema": "s",
649 "item": "src",
650 "size": "small",
651 "type": "kafka"
652 }"#,
653 );
654 }
655
656 #[mz_ore::test]
662 fn rename_hydration_time_estimate_under_scheduling() {
663 check(
664 r#"{"CreateClusterReplicaV3": {
665 "cluster_id": "u1",
666 "cluster_name": "c",
667 "replica_id": {"inner": "r1"},
668 "replica_name": "n",
669 "logical_size": "small",
670 "disk": false,
671 "billed_as": null,
672 "internal": false,
673 "reason": {"reason": {"Manual": {}}},
674 "scheduling_policies": {
675 "on_refresh": {
676 "decision": {"On": {}},
677 "objects_needing_refresh": [],
678 "objects_needing_compaction": [],
679 "rehydration_time_estimate": "00:00:07"
680 }
681 }
682 }}"#,
683 r#"{
684 "cluster_id": "u1",
685 "cluster_name": "c",
686 "replica_id": "r1",
687 "replica_name": "n",
688 "logical_size": "small",
689 "disk": false,
690 "billed_as": null,
691 "internal": false,
692 "reason": "manual",
693 "scheduling_policies": {
694 "on_refresh": {
695 "decision": "on",
696 "objects_needing_refresh": [],
697 "objects_needing_compaction": [],
698 "hydration_time_estimate": "00:00:07"
699 }
700 }
701 }"#,
702 );
703 }
704
705 #[mz_ore::test]
709 fn drop_null_replica_id_v1() {
710 check(
711 r#"{"CreateClusterReplicaV1": {
712 "cluster_id": "u1",
713 "cluster_name": "c",
714 "replica_id": null,
715 "replica_name": "n",
716 "logical_size": "small",
717 "disk": false,
718 "billed_as": null,
719 "internal": false
720 }}"#,
721 r#"{
722 "cluster_id": "u1",
723 "cluster_name": "c",
724 "replica_name": "n",
725 "logical_size": "small",
726 "disk": false,
727 "billed_as": null,
728 "internal": false
729 }"#,
730 );
731 }
732
733 #[mz_ore::test]
738 fn drop_null_replica_id_v2_kept() {
739 check(
740 r#"{"CreateClusterReplicaV2": {
741 "cluster_id": "u1",
742 "cluster_name": "c",
743 "replica_id": null,
744 "replica_name": "n",
745 "logical_size": "small",
746 "disk": false,
747 "billed_as": null,
748 "internal": false,
749 "reason": {"reason": {"System": {}}},
750 "scheduling_policies": null
751 }}"#,
752 r#"{
753 "cluster_id": "u1",
754 "cluster_name": "c",
755 "replica_id": null,
756 "replica_name": "n",
757 "logical_size": "small",
758 "disk": false,
759 "billed_as": null,
760 "internal": false,
761 "reason": "system"
762 }"#,
763 );
764 }
765
766 #[mz_ore::test]
769 fn enum_single_wrap_decision() {
770 check(
771 r#"{"CreateClusterReplicaV2": {
772 "cluster_id": "u1",
773 "cluster_name": "c",
774 "replica_id": {"inner": "r1"},
775 "replica_name": "n",
776 "logical_size": "small",
777 "disk": false,
778 "billed_as": null,
779 "internal": false,
780 "reason": {"reason": {"Schedule": {}}},
781 "scheduling_policies": {
782 "on_refresh": {
783 "decision": {"Off": {}},
784 "objects_needing_refresh": [],
785 "rehydration_time_estimate": "00:00:00"
786 }
787 }
788 }}"#,
789 r#"{
790 "cluster_id": "u1",
791 "cluster_name": "c",
792 "replica_id": "r1",
793 "replica_name": "n",
794 "logical_size": "small",
795 "disk": false,
796 "billed_as": null,
797 "internal": false,
798 "reason": "schedule",
799 "scheduling_policies": {
800 "on_refresh": {
801 "decision": "off",
802 "objects_needing_refresh": [],
803 "hydration_time_estimate": "00:00:00"
804 }
805 }
806 }"#,
807 );
808 }
809
810 #[mz_ore::test]
814 fn enum_double_wrap_reason_kebab() {
815 check(
816 r#"{"DropClusterReplicaV2": {
817 "cluster_id": "u1",
818 "cluster_name": "c",
819 "replica_id": "r1",
820 "replica_name": "n",
821 "reason": {"reason": {"HydrationBurst": {}}},
822 "scheduling_policies": null
823 }}"#,
824 r#"{
825 "cluster_id": "u1",
826 "cluster_name": "c",
827 "replica_id": "r1",
828 "replica_name": "n",
829 "reason": "hydration-burst"
830 }"#,
831 );
832 }
833
834 #[mz_ore::test]
838 #[cfg_attr(miri, ignore)] fn enum_double_wrap_transition_timed_out() {
840 check(
841 r#"{"AlterClusterReconfigurationV1": {
842 "cluster_id": "u1",
843 "cluster_name": "c",
844 "transition": {"transition": {"TimedOut": {}}},
845 "target_size": "small",
846 "target_replication_factor": 1,
847 "target_availability_zones": [],
848 "target_logging": {"log_logging": false, "interval": null},
849 "deadline": null
850 }}"#,
851 r#"{
852 "cluster_id": "u1",
853 "cluster_name": "c",
854 "transition": "timed-out",
855 "target_size": "small",
856 "target_replication_factor": 1,
857 "target_availability_zones": [],
858 "target_logging": {"log_logging": false, "interval": null},
859 "deadline": null
860 }"#,
861 );
862 }
863
864 #[mz_ore::test]
869 fn enum_double_wrap_hydration_burst_finished() {
870 check(
871 r#"{"ClusterHydrationBurstV1": {
872 "cluster_id": "u1",
873 "cluster_name": "c",
874 "transition": {"transition": {"Finished": {}}},
875 "burst_size": "small"
876 }}"#,
877 r#"{
878 "cluster_id": "u1",
879 "cluster_name": "c",
880 "transition": "finished",
881 "burst_size": "small"
882 }"#,
883 );
884 }
885
886 #[mz_ore::test]
890 fn reset_all_v1_is_null() {
891 check(r#"{"ResetAllV1": {}}"#, r#"null"#);
892 }
893
894 #[mz_ore::test]
898 fn enum_unknown_variant_errors() {
899 check_err(
900 r#"{"DropClusterReplicaV2": {
901 "cluster_id": "u1",
902 "cluster_name": "c",
903 "replica_id": "r1",
904 "replica_name": "n",
905 "reason": {"reason": {"NoSuchVariant": {}}},
906 "scheduling_policies": null
907 }}"#,
908 "unknown enum variant NoSuchVariant",
909 );
910 }
911
912 #[mz_ore::test]
916 fn enum_non_empty_payload_errors() {
917 check_err(
918 r#"{"DropClusterReplicaV2": {
919 "cluster_id": "u1",
920 "cluster_name": "c",
921 "replica_id": "r1",
922 "replica_name": "n",
923 "reason": {"reason": {"Manual": {"unexpected": "x"}}},
924 "scheduling_policies": null
925 }}"#,
926 "non-empty payload",
927 );
928 }
929
930 #[mz_ore::test]
932 #[cfg_attr(miri, ignore)] fn error_cases() {
934 check_err(r#"{}"#, "empty details enum");
935 check_err(
936 r#"{"A": {"x": 1}, "B": {"y": 2}}"#,
937 "details enum had multiple keys",
938 );
939 check_err(r#"["IdNameV1", {"id": "u1"}]"#, "expected object");
940 check_err(r#"{"IdNameV1": "not an object"}"#, "expected inner object");
941 }
942}