End-to-end Glue source tests + kafka-ingest glue-wire-format

Author: martykulma

ci/nightly/pipeline.template.yml

@@ -949,6 +949,23 @@ steps:
           - ./ci/plugins/mzcompose:
               composition: aws-localstack
 
+      - id: aws-glue-schema-registry-real
+        label: AWS Glue Schema Registry (Real)
+        depends_on: build-aarch64
+        timeout_in_minutes: 30
+        retry:
+          automatic:
+            - exit_status: 1
+              limit: 1
+        agents:
+          # Because of scratch-aws-access
+          queue: linux-aarch64-small
+        plugins:
+          - ./ci/plugins/scratch-aws-access: ~
+          - ./ci/plugins/mzcompose:
+              composition: aws-glue-schema-registry
+              run: aws
+
       - id: secrets-local-file
         label: "Secrets Local File"
         depends_on: build-aarch64

ci/test/lint-main/checks/check-mzcompose-files.sh

@@ -50,6 +50,7 @@ check_default_workflow_references_others() {
         -not -wholename "./test/cluster-spec-sheet/mzcompose.py" `# Handled differently` \
         -not -wholename "./test/orchestratord/mzcompose.py" `# Handled differently` \
         -not -wholename "./test/workload-replay/mzcompose.py" `# Handled differently` \
+        -not -wholename "./test/aws-glue-schema-registry/mzcompose.py" `# 'aws' workflow runs against real AWS, opt-in via nightly only` \
     )
 
     for file in "${MZCOMPOSE_TEST_FILES[@]}"; do

ci/test/pipeline.template.yml

@@ -467,6 +467,17 @@ steps:
         agents:
           queue: hetzner-aarch64-4cpu-8gb
 
+      - id: aws-glue-schema-registry
+        label: AWS Glue Schema Registry (Moto)
+        depends_on: build-aarch64
+        timeout_in_minutes: 30
+        artifact_paths: junit_*.xml
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: aws-glue-schema-registry
+        agents:
+          queue: hetzner-aarch64-4cpu-8gb
+
   - id: zippy-kafka-sources-short
     label: "Short Zippy"
     depends_on: build-aarch64

misc/python/materialize/mzcompose/services/moto.py

@@ -0,0 +1,44 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+
+from materialize.mzcompose.service import (
+    Service,
+)
+
+
+class Moto(Service):
+    """A `moto_server` container that mocks AWS services.
+
+    Used for testing AWS surfaces that LocalStack Community doesn't cover
+    (notably Glue / Glue Schema Registry).
+    """
+
+    def __init__(
+        self,
+        name: str = "moto",
+        image: str = "motoserver/moto:5.2.1",
+        port: int = 5000,
+    ) -> None:
+        super().__init__(
+            name=name,
+            config={
+                "image": image,
+                "init": True,
+                "ports": [port],
+                "healthcheck": {
+                    "test": [
+                        "CMD-SHELL",
+                        f"python -c 'import urllib.request; urllib.request.urlopen(\"http://localhost:{port}/moto-api/\")'",
+                    ],
+                    "interval": "1s",
+                    "start_period": "30s",
+                },
+            },
+        )

src/interchange/src/avro/schema.rs

@@ -549,8 +549,24 @@ impl fmt::Debug for AvroSchemaResolver {
 ///
 /// Differences from the CSR cache:
 /// * Keys are UUIDs (Glue schema-version IDs), not `i32`s.
-/// * Glue schemas are single definitions — no `references` field on
-///   `GetSchemaVersion`, so the cache does not chase a dependency graph.
+/// * **No cross-schema references.** AWS Glue Schema Registry does not have
+///   a CSR-style "schema references" concept: `GetSchemaVersion` returns a
+///   single self-contained `SchemaDefinition` JSON blob with no `references`
+///   field, so the cache does not chase a dependency graph (compare
+///   [`SchemaCache::get`], which fetches transitive references). Avro's
+///   *intra-document* named-type references still work — they are resolved
+///   by the Avro parser from the single JSON document.
+/// * **Registry name enforcement.** Glue schema-version UUIDs are globally
+///   unique within an AWS account, and `GetSchemaVersion(uuid)` is not
+///   scoped to a registry. The AWS API forces a choice between
+///   `SchemaVersionId` (UUID) and `SchemaId(RegistryName, SchemaName) +
+///   SchemaVersionNumber`; combining them is not allowed (see
+///   <https://docs.aws.amazon.com/glue/latest/webapi/API_GetSchemaVersion.html>).
+///   At runtime we only have the UUID, so the cache validates each
+///   fetched version's `SchemaArn` against `expected_registry` and
+///   treats a mismatch as a permanent decode error — otherwise a record
+///   framed against the wrong registry would silently succeed as long as
+///   the credentials could see it.
 /// * No outer retry layer. `aws-sdk-glue` ships a "standard" retry policy
 ///   by default that handles transient errors; layering our own
 ///   `Retry::default()` on top would amplify backoff without adding

src/sql-parser/src/ast/defs/ddl.rs

@@ -174,6 +174,15 @@ pub enum AvroSchema<T: AstInfo> {
     ///
     /// Parallel to the `Csr` variant. See
     /// `doc/developer/design/20260512_aws_glue_schema_registry.md`.
+    ///
+    /// **Cross-schema references are intentionally unsupported.** Unlike the
+    /// `Csr` variant — whose `key_reference_schemas` / `value_reference_schemas`
+    /// fields are resolved transitively against the Confluent registry — AWS
+    /// Glue Schema Registry has no analogue: `RegisterSchemaVersion` takes a
+    /// single self-contained `SchemaDefinition` JSON blob with no
+    /// cross-schema-reference field. Avro's *intra-document* named-type
+    /// references still work because they are resolved by the Avro parser
+    /// from a single JSON document.
     Glue {
         connection: T::ItemName,
         with_options: Vec<GlueAvroOption<T>>,

src/sql/src/plan/statement/ddl.rs

@@ -2415,6 +2415,12 @@ fn get_encoding_inner(
                         sql_bail!("Avro Glue seed resolution has not been performed");
                     };
 
+                    // A Glue seed carries a single `value_schema` (unlike a
+                    // CSR seed, which can hold both key and value), so a bare
+                    // `FORMAT AVRO USING AWS GLUE` is always value-only. A
+                    // key is still expressible via the separate `KEY FORMAT
+                    // ... VALUE FORMAT ...` clause, where `get_encoding`
+                    // promotes this value encoding into the key slot.
                     Schema {
                         key_schema: None,
                         value_schema: seed.value_schema.clone(),

src/testdrive/src/action/kafka/ingest.rs

@@ -21,6 +21,7 @@ use rdkafka::message::{Header, OwnedHeaders};
 use rdkafka::producer::FutureRecord;
 use serde::de::DeserializeOwned;
 use tokio::fs;
+use uuid::Uuid;
 
 use crate::action::{self, ControlFlow, State};
 use crate::format::avro::{self, Schema};
@@ -160,6 +161,10 @@ enum Format {
     Avro {
         schema: String,
         confluent_wire_format: bool,
+        /// If set, override the wire format with AWS Glue Schema Registry framing
+        /// using the given schema-version UUID. Mutually exclusive with
+        /// `confluent_wire_format=true`.
+        glue_schema_version_id: Option<Uuid>,
         /// Schema references (subject names) for Confluent Schema Registry
         references: Vec<String>,
     },
@@ -183,6 +188,10 @@ enum Transcoder {
         schema: Schema,
         schema_id: i32,
     },
+    GlueAvro {
+        schema: Schema,
+        schema_version_id: Uuid,
+    },
     Protobuf {
         message: MessageDescriptor,
         confluent_wire_format: bool,
@@ -230,6 +239,26 @@ impl Transcoder {
                     Ok(None)
                 }
             }
+            Transcoder::GlueAvro {
+                schema,
+                schema_version_id,
+            } => {
+                if let Some(val) = Self::decode_json(row)? {
+                    let val = avro::from_json(&val, schema.top_node())?;
+                    let mut out = vec![];
+                    // AWS Glue Schema Registry wire format:
+                    // byte 0   = 0x03 (header version)
+                    // byte 1   = compression byte (0x00 = none)
+                    // bytes 2..18 = 16-byte schema-version UUID
+                    out.write_u8(0x03).unwrap();
+                    out.write_u8(0x00).unwrap();
+                    out.extend_from_slice(schema_version_id.as_bytes());
+                    out.extend(avro::to_avro_datum(schema, val)?);
+                    Ok(Some(out))
+                } else {
+                    Ok(None)
+                }
+            }
             Transcoder::PlainAvro { schema } => {
                 if let Some(val) = Self::decode_json(row)? {
                     let val = avro::from_json(&val, schema.top_node())?;
@@ -303,16 +332,31 @@ pub async fn run_ingest(
     let schema_id_var = cmd.args.opt_parse("set-schema-id-var")?;
     let key_schema_id_var = cmd.args.opt_parse("set-key-schema-id-var")?;
     let format = match cmd.args.string("format")?.as_str() {
-        "avro" => Format::Avro {
-            schema: cmd.args.string("schema")?,
-            confluent_wire_format: cmd.args.opt_bool("confluent-wire-format")?.unwrap_or(true),
-            // TODO (maz): update README!
-            references: cmd
+        "avro" => {
+            let glue_schema_version_id = cmd
                 .args
-                .opt_string("references")
-                .map(|s| s.split(',').map(|s| s.to_string()).collect())
-                .unwrap_or_default(),
-        },
+                .opt_string("glue-schema-version-id")
+                .map(|s| Uuid::parse_str(&s).context("parsing glue-schema-version-id as UUID"))
+                .transpose()?;
+            let confluent_wire_format_explicit = cmd.args.opt_bool("confluent-wire-format")?;
+            if glue_schema_version_id.is_some() && confluent_wire_format_explicit == Some(true) {
+                bail!("confluent-wire-format=true is incompatible with glue-schema-version-id");
+            }
+            // Default: confluent unless Glue framing is requested.
+            let confluent_wire_format =
+                confluent_wire_format_explicit.unwrap_or_else(|| glue_schema_version_id.is_none());
+            Format::Avro {
+                schema: cmd.args.string("schema")?,
+                confluent_wire_format,
+                glue_schema_version_id,
+                // TODO (maz): update README!
+                references: cmd
+                    .args
+                    .opt_string("references")
+                    .map(|s| s.split(',').map(|s| s.to_string()).collect())
+                    .unwrap_or_default(),
+            }
+        }
         "protobuf" => {
             let descriptor_file = cmd.args.string("descriptor-file")?;
             let message = cmd.args.string("message")?;
@@ -336,6 +380,7 @@ pub async fn run_ingest(
                 anyhow!("key-schema parameter required when key-format is present")
             })?,
             confluent_wire_format: cmd.args.opt_bool("confluent-wire-format")?.unwrap_or(true),
+            glue_schema_version_id: None,
             references: cmd
                 .args
                 .opt_string("key-references")
@@ -425,8 +470,16 @@ pub async fn run_ingest(
             match fmt {
                 Format::Avro {
                     confluent_wire_format,
+                    glue_schema_version_id,
                     ..
-                } => Some(*confluent_wire_format),
+                } => {
+                    // Glue framing is its own wire format — don't compare against CSR.
+                    if glue_schema_version_id.is_some() {
+                        None
+                    } else {
+                        Some(*confluent_wire_format)
+                    }
+                }
                 Format::Protobuf {
                     confluent_wire_format,
                     ..
@@ -550,8 +603,20 @@ async fn make_transcoder(
         Format::Avro {
             schema,
             confluent_wire_format,
+            glue_schema_version_id,
             references,
         } => {
+            if let Some(schema_version_id) = glue_schema_version_id {
+                if !references.is_empty() {
+                    bail!("schema references are not supported with glue-schema-version-id");
+                }
+                let schema = avro::parse_schema(&schema, &[])
+                    .with_context(|| format!("parsing avro schema: {}", schema))?;
+                return Ok(Transcoder::GlueAvro {
+                    schema,
+                    schema_version_id,
+                });
+            }
             if confluent_wire_format {
                 // Build references list by fetching each subject from the registry.
                 // Start with immediate references and automatically resolve transitive ones.