mz_interchange/glue.rs
1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Wire-format helpers for the AWS Glue Schema Registry framing.
11//!
12//! Glue prepends an 18-byte header to each Kafka record payload:
13//!
14//! | Offset | Bytes | Meaning |
15//! |--------|-------|------------------------------------------------------|
16//! | 0 | 1 | Header version. Glue currently emits `0x03`. |
17//! | 1 | 1 | Compression byte. `0x00` = none, `0x05` = zlib. |
18//! | 2..18 | 16 | Schema-version UUID, big-endian. |
19//! | 18.. | N | The serialized record payload. |
20//!
21//! Not documented in any spec (that I could find it), ref to the source
22//! [serializer-deserializer/src/main/java/com/amazonaws/services/schemaregistry/serializers/SerializationDataEncoder.java](https://github.com/awslabs/aws-glue-schema-registry/blob/4b9cac477d6876a883e2a8893738a30c072694dc/serializer-deserializer/src/main/java/com/amazonaws/services/schemaregistry/serializers/SerializationDataEncoder.java#L54-L70)
23//! Materialize only supports the uncompressed framing (`compression =
24//! 0x00`). Compressed records are rejected — supporting zlib at the wire
25//! layer is straightforward but no consumer in Materialize asks for it yet,
26//! and silently decompressing would mask producer misconfiguration.
27//!
28//! The Confluent analogue lives in [`crate::confluent`].
29
30use anyhow::{Result, bail};
31use uuid::Uuid;
32
33/// Glue wire-format header version, written at byte 0.
34const HEADER_VERSION: u8 = 0x03;
35
36/// Compression byte indicating an uncompressed payload.
37const COMPRESSION_NONE: u8 = 0x00;
38
39/// Length of the Glue header in bytes (version + compression + UUID).
40pub const HEADER_LEN: usize = 1 + 1 + 16;
41
42/// Parse the Glue Avro header from the front of `buf`, returning the
43/// schema-version UUID and a subslice covering the record payload.
44///
45/// Returns an error if the buffer is shorter than the fixed header, if the
46/// header-version byte is not `0x03`, or if the compression byte is
47/// anything other than `0x00`.
48pub fn extract_avro_header(buf: &[u8]) -> Result<(Uuid, &[u8])> {
49 if buf.len() < HEADER_LEN {
50 bail!(
51 "Glue-style avro datum is too few bytes: expected at least {} bytes, got {}",
52 HEADER_LEN,
53 buf.len()
54 );
55 }
56 let version = buf[0];
57 if version != HEADER_VERSION {
58 bail!(
59 "wrong Glue-style avro serialization header version: expected {:#04x}, got {:#04x}",
60 HEADER_VERSION,
61 version
62 );
63 }
64 let compression = buf[1];
65 if compression != COMPRESSION_NONE {
66 bail!(
67 "unsupported Glue-style avro compression byte: \
68 expected {:#04x} (uncompressed), got {:#04x}",
69 COMPRESSION_NONE,
70 compression
71 );
72 }
73 // `Uuid::from_slice` only fails on length mismatch, which we've already
74 // validated above; the unwrap is sound.
75 let uuid = Uuid::from_slice(&buf[2..HEADER_LEN]).expect("18-byte header validated above");
76 Ok((uuid, &buf[HEADER_LEN..]))
77}
78
79/// Write the Glue Avro header to `buf`, using the uncompressed framing
80/// (`compression = 0x00`). Callers append the serialized record payload
81/// directly after the header, so the framed record needs only a single
82/// allocation.
83pub fn write_avro_header(buf: &mut Vec<u8>, schema_version_id: Uuid) {
84 buf.reserve(HEADER_LEN);
85 buf.push(HEADER_VERSION);
86 buf.push(COMPRESSION_NONE);
87 buf.extend_from_slice(schema_version_id.as_bytes());
88}
89
90#[cfg(test)]
91mod tests {
92 use super::*;
93
94 fn fixture_uuid() -> Uuid {
95 // Fixed value so the encoded byte layout is exact in assertions.
96 Uuid::parse_str("12345678-1234-5678-1234-567812345678").unwrap()
97 }
98
99 fn framed(uuid: Uuid, payload: &[u8]) -> Vec<u8> {
100 let mut buf = Vec::new();
101 write_avro_header(&mut buf, uuid);
102 buf.extend_from_slice(payload);
103 buf
104 }
105
106 #[mz_ore::test]
107 fn roundtrip() {
108 let uuid = fixture_uuid();
109 let payload = b"avro-bytes-here";
110 let framed = framed(uuid, payload);
111 assert_eq!(framed.len(), HEADER_LEN + payload.len());
112 let (parsed_uuid, rest) = extract_avro_header(&framed).unwrap();
113 assert_eq!(parsed_uuid, uuid);
114 assert_eq!(rest, payload);
115 }
116
117 #[mz_ore::test]
118 fn header_byte_layout() {
119 let uuid = fixture_uuid();
120 let framed = framed(uuid, &[]);
121 assert_eq!(framed[0], HEADER_VERSION);
122 assert_eq!(framed[1], COMPRESSION_NONE);
123 assert_eq!(&framed[2..HEADER_LEN], uuid.as_bytes());
124 }
125
126 #[mz_ore::test]
127 fn rejects_buffer_too_short() {
128 // 17 bytes — one short of the minimum header.
129 let buf = [0u8; HEADER_LEN - 1];
130 let err = extract_avro_header(&buf).unwrap_err();
131 assert!(err.to_string().contains("too few bytes"), "{err}");
132 }
133
134 #[mz_ore::test]
135 fn rejects_wrong_header_version() {
136 let mut buf = framed(fixture_uuid(), b"payload");
137 buf[0] = 0x02;
138 let err = extract_avro_header(&buf).unwrap_err();
139 assert!(
140 err.to_string()
141 .contains("wrong Glue-style avro serialization header version"),
142 "{err}"
143 );
144 }
145
146 #[mz_ore::test]
147 fn rejects_compressed_payload() {
148 let mut buf = framed(fixture_uuid(), b"payload");
149 buf[1] = 0x05; // zlib
150 let err = extract_avro_header(&buf).unwrap_err();
151 assert!(
152 err.to_string()
153 .contains("unsupported Glue-style avro compression byte"),
154 "{err}"
155 );
156 }
157
158 #[mz_ore::test]
159 fn empty_payload_is_legal() {
160 let uuid = fixture_uuid();
161 let framed = framed(uuid, &[]);
162 let (parsed_uuid, rest) = extract_avro_header(&framed).unwrap();
163 assert_eq!(parsed_uuid, uuid);
164 assert!(rest.is_empty());
165 }
166}