SS-16 AWS Glue Support: SQL

src/adapter/src/catalog/builtin_table_updates.rs

@@ -938,6 +938,7 @@ impl CatalogState {
                 updates.push(self.pack_ssh_tunnel_connection_update(id, key_1, key_2, diff));
             }
             ConnectionDetails::Csr(_)
+            | ConnectionDetails::GlueSchemaRegistry(_)
             | ConnectionDetails::Postgres(_)
             | ConnectionDetails::MySql(_)
             | ConnectionDetails::SqlServer(_)

src/adapter/src/catalog/state.rs

@@ -2772,6 +2772,7 @@ impl ConnectionResolver for CatalogState {
             Kafka(conn) => Kafka(conn.into_inline_connection(self)),
             Postgres(conn) => Postgres(conn.into_inline_connection(self)),
             Csr(conn) => Csr(conn.into_inline_connection(self)),
+            GlueSchemaRegistry(conn) => GlueSchemaRegistry(conn.into_inline_connection(self)),
             Ssh(conn) => Ssh(conn),
             Aws(conn) => Aws(conn),
             AwsPrivatelink(conn) => AwsPrivatelink(conn),

src/adapter/src/coord/ddl.rs

@@ -1133,6 +1133,7 @@ impl Coordinator {
                                 new_aws_privatelink_connections += 1
                             }
                             ConnectionDetails::Csr(_)
+                            | ConnectionDetails::GlueSchemaRegistry(_)
                             | ConnectionDetails::Ssh { .. }
                             | ConnectionDetails::Aws(_)
                             | ConnectionDetails::IcebergCatalog(_) => {}
@@ -1308,6 +1309,7 @@ impl Coordinator {
                 ConnectionDetails::SqlServer(_) => current_sql_server_connections += 1,
                 ConnectionDetails::Kafka(_) => current_kafka_connections += 1,
                 ConnectionDetails::Csr(_)
+                | ConnectionDetails::GlueSchemaRegistry(_)
                 | ConnectionDetails::Ssh { .. }
                 | ConnectionDetails::Aws(_)
                 | ConnectionDetails::IcebergCatalog(_) => {}

src/sql-lexer/src/keywords.txt

@@ -201,6 +201,7 @@ Fullname
 Function
 Fusion
 Generator
+Glue
 Grant
 Greatest
 Group

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

@@ -785,6 +785,7 @@ pub enum ConnectionOptionName {
     PublicKey1,
     PublicKey2,
     Region,
+    Registry,
     SaslMechanisms,
     SaslPassword,
     SaslUsername,
@@ -826,6 +827,7 @@ impl AstDisplay for ConnectionOptionName {
             ConnectionOptionName::PublicKey1 => "PUBLIC KEY 1",
             ConnectionOptionName::PublicKey2 => "PUBLIC KEY 2",
             ConnectionOptionName::Region => "REGION",
+            ConnectionOptionName::Registry => "REGISTRY",
             ConnectionOptionName::AssumeRoleArn => "ASSUME ROLE ARN",
             ConnectionOptionName::AssumeRoleSessionName => "ASSUME ROLE SESSION NAME",
             ConnectionOptionName::SaslMechanisms => "SASL MECHANISMS",
@@ -875,6 +877,7 @@ impl WithOptionName for ConnectionOptionName {
             | ConnectionOptionName::PublicKey1
             | ConnectionOptionName::PublicKey2
             | ConnectionOptionName::Region
+            | ConnectionOptionName::Registry
             | ConnectionOptionName::AssumeRoleArn
             | ConnectionOptionName::AssumeRoleSessionName
             | ConnectionOptionName::SaslMechanisms
@@ -911,6 +914,7 @@ impl_display_t!(ConnectionOption);
 pub enum CreateConnectionType {
     Aws,
     AwsPrivatelink,
+    GlueSchemaRegistry,
     Kafka,
     Csr,
     Postgres,
@@ -928,6 +932,7 @@ impl CreateConnectionType {
             Self::Postgres => "postgres",
             Self::Aws => "aws",
             Self::AwsPrivatelink => "aws-privatelink",
+            Self::GlueSchemaRegistry => "glue-schema-registry",
             Self::Ssh => "ssh-tunnel",
             Self::MySql => "mysql",
             Self::SqlServer => "sql-server",
@@ -954,6 +959,9 @@ impl AstDisplay for CreateConnectionType {
             Self::AwsPrivatelink => {
                 f.write_str("AWS PRIVATELINK");
             }
+            Self::GlueSchemaRegistry => {
+                f.write_str("AWS GLUE SCHEMA REGISTRY");
+            }
             Self::Ssh => {
                 f.write_str("SSH TUNNEL");
             }

src/sql-parser/src/parser.rs

@@ -2462,6 +2462,9 @@ impl<'a> Parser<'a> {
             AWS => {
                 if self.parse_keyword(PRIVATELINK) {
                     CreateConnectionType::AwsPrivatelink
+                } else if self.parse_keyword(GLUE) {
+                    self.expect_keywords(&[SCHEMA, REGISTRY])?;
+                    CreateConnectionType::GlueSchemaRegistry
                 } else {
                     CreateConnectionType::Aws
                 }
@@ -2792,6 +2795,7 @@ impl<'a> Parser<'a> {
                 PUBLIC,
                 PROGRESS,
                 REGION,
+                REGISTRY,
                 ROLE,
                 SASL,
                 SCOPE,
@@ -2862,6 +2866,7 @@ impl<'a> Parser<'a> {
                     ConnectionOptionName::SecurityProtocol
                 }
                 REGION => ConnectionOptionName::Region,
+                REGISTRY => ConnectionOptionName::Registry,
                 SASL => match self.expect_one_of_keywords(&[MECHANISMS, PASSWORD, USERNAME])? {
                     MECHANISMS => ConnectionOptionName::SaslMechanisms,
                     PASSWORD => ConnectionOptionName::SaslPassword,

src/sql-parser/tests/testdata/ddl

@@ -589,6 +589,27 @@ CREATE CONNECTION icebergcatalog TO ICEBERG CATALOG (CATALOG TYPE = 's3tablesres
 =>
 CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("icebergcatalog")]), connection_type: IcebergCatalog, if_not_exists: false, values: [ConnectionOption { name: CatalogType, value: Some(Value(String("s3tablesrest"))) }, ConnectionOption { name: AwsConnection, value: Some(Item(Name(UnresolvedItemName([Ident("awsconn")])))) }, ConnectionOption { name: Warehouse, value: Some(Value(String("wh"))) }], with_options: [] })
 
+parse-statement
+CREATE CONNECTION glueconn TO AWS GLUE SCHEMA REGISTRY (AWS CONNECTION = awsconn, REGISTRY = 'my-registry')
+----
+CREATE CONNECTION glueconn TO AWS GLUE SCHEMA REGISTRY (AWS CONNECTION = awsconn, REGISTRY = 'my-registry')
+=>
+CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("glueconn")]), connection_type: GlueSchemaRegistry, if_not_exists: false, values: [ConnectionOption { name: AwsConnection, value: Some(Item(Name(UnresolvedItemName([Ident("awsconn")])))) }, ConnectionOption { name: Registry, value: Some(Value(String("my-registry"))) }], with_options: [] })
+
+parse-statement
+CREATE CONNECTION glueconn FOR AWS GLUE SCHEMA REGISTRY AWS CONNECTION awsconn, REGISTRY 'my-registry'
+----
+CREATE CONNECTION glueconn TO AWS GLUE SCHEMA REGISTRY (AWS CONNECTION = awsconn, REGISTRY = 'my-registry')
+=>
+CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("glueconn")]), connection_type: GlueSchemaRegistry, if_not_exists: false, values: [ConnectionOption { name: AwsConnection, value: Some(Item(Name(UnresolvedItemName([Ident("awsconn")])))) }, ConnectionOption { name: Registry, value: Some(Value(String("my-registry"))) }], with_options: [] })
+
+parse-statement
+CREATE CONNECTION glueconn TO AWS GLUE SCHEMA REGISTRY (AWS CONNECTION = awsconn, REGISTRY = 'my-registry') WITH (VALIDATE = FALSE)
+----
+CREATE CONNECTION glueconn TO AWS GLUE SCHEMA REGISTRY (AWS CONNECTION = awsconn, REGISTRY = 'my-registry') WITH (VALIDATE = false)
+=>
+CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("glueconn")]), connection_type: GlueSchemaRegistry, if_not_exists: false, values: [ConnectionOption { name: AwsConnection, value: Some(Item(Name(UnresolvedItemName([Ident("awsconn")])))) }, ConnectionOption { name: Registry, value: Some(Value(String("my-registry"))) }], with_options: [CreateConnectionOption { name: Validate, value: Some(Value(Boolean(false))) }] })
+
 parse-statement
 CREATE CONNECTION pgconn FOR postgres HOST foo, PORT 1234, SSL CERTIFICATE AUTHORITY 'foo', SSH TUNNEL tun, DATABASE 'db', PASSWORD 'pw', SSL CERTIFICATE 'cert', SSL KEY 'key', SSL MODE 'mode', USER 'postgres'
 ----
@@ -2820,7 +2841,7 @@ CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("my
 parse-statement
 CREATE CONNECTION my_ssh_tunnel FOR SSH TUNNEL (PUBLIC KEY 3 = nope)
 ----
-error: Expected one of ACCESS or ASSUME or AVAILABILITY or AWS or BROKER or BROKERS or CATALOG or CREDENTIAL or DATABASE or ENDPOINT or HOST or PASSWORD or PORT or PUBLIC or PROGRESS or REGION or ROLE or SASL or SCOPE or SECRET or SECURITY or SERVICE or SESSION or SSH or SSL or URL or USER or USERNAME or WAREHOUSE, found left parenthesis
+error: Expected one of ACCESS or ASSUME or AVAILABILITY or AWS or BROKER or BROKERS or CATALOG or CREDENTIAL or DATABASE or ENDPOINT or HOST or PASSWORD or PORT or PUBLIC or PROGRESS or REGION or REGISTRY or ROLE or SASL or SCOPE or SECRET or SECURITY or SERVICE or SESSION or SSH or SSL or URL or USER or USERNAME or WAREHOUSE, found left parenthesis
 CREATE CONNECTION my_ssh_tunnel FOR SSH TUNNEL (PUBLIC KEY 3 = nope)
                                                ^
 

src/sql/src/plan.rs

@@ -59,8 +59,9 @@ use mz_ssh_util::keys::SshKeyPair;
 use mz_storage_types::connections::aws::AwsConnection;
 use mz_storage_types::connections::inline::ReferencedConnection;
 use mz_storage_types::connections::{
-    AwsPrivatelinkConnection, CsrConnection, IcebergCatalogConnection, KafkaConnection,
-    MySqlConnection, PostgresConnection, SqlServerConnectionDetails, SshConnection,
+    AwsPrivatelinkConnection, CsrConnection, GlueSchemaRegistryConnection,
+    IcebergCatalogConnection, KafkaConnection, MySqlConnection, PostgresConnection,
+    SqlServerConnectionDetails, SshConnection,
 };
 use mz_storage_types::instances::StorageInstanceId;
 use mz_storage_types::sinks::{S3SinkFormat, SinkEnvelope, StorageSinkConnection};
@@ -1668,6 +1669,7 @@ pub struct Connection {
 pub enum ConnectionDetails {
     Kafka(KafkaConnection<ReferencedConnection>),
     Csr(CsrConnection<ReferencedConnection>),
+    GlueSchemaRegistry(GlueSchemaRegistryConnection<ReferencedConnection>),
     Postgres(PostgresConnection<ReferencedConnection>),
     Ssh {
         connection: SshConnection,
@@ -1688,6 +1690,9 @@ impl ConnectionDetails {
                 mz_storage_types::connections::Connection::Kafka(c.clone())
             }
             ConnectionDetails::Csr(c) => mz_storage_types::connections::Connection::Csr(c.clone()),
+            ConnectionDetails::GlueSchemaRegistry(c) => {
+                mz_storage_types::connections::Connection::GlueSchemaRegistry(c.clone())
+            }
             ConnectionDetails::Postgres(c) => {
                 mz_storage_types::connections::Connection::Postgres(c.clone())
             }

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

@@ -7066,6 +7066,7 @@ pub fn plan_alter_connection(
         Connection::AwsPrivatelink(_) => CreateConnectionType::AwsPrivatelink,
         Connection::Kafka(_) => CreateConnectionType::Kafka,
         Connection::Csr(_) => CreateConnectionType::Csr,
+        Connection::GlueSchemaRegistry(_) => CreateConnectionType::GlueSchemaRegistry,
         Connection::Postgres(_) => CreateConnectionType::Postgres,
         Connection::Ssh(_) => CreateConnectionType::Ssh,
         Connection::MySql(_) => CreateConnectionType::MySql,

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

@@ -33,10 +33,11 @@ use mz_storage_types::connections::inline::ReferencedConnection;
 use mz_storage_types::connections::string_or_secret::StringOrSecret;
 use mz_storage_types::connections::{
     AwsPrivatelink, AwsPrivatelinkConnection, AwsPrivatelinkRule, CsrConnection,
-    CsrConnectionHttpAuth, IcebergCatalogConnection, IcebergCatalogImpl, IcebergCatalogType,
-    KafkaConnection, KafkaSaslConfig, KafkaTlsConfig, KafkaTopicOptions, MySqlConnection,
-    MySqlSslMode, PostgresConnection, RestIcebergCatalog, S3TablesRestIcebergCatalog,
-    SqlServerConnectionDetails, SshConnection, SshTunnel, TlsIdentity, Tunnel,
+    CsrConnectionHttpAuth, GlueSchemaRegistryConnection, IcebergCatalogConnection,
+    IcebergCatalogImpl, IcebergCatalogType, KafkaConnection, KafkaSaslConfig, KafkaTlsConfig,
+    KafkaTopicOptions, MySqlConnection, MySqlSslMode, PostgresConnection, RestIcebergCatalog,
+    S3TablesRestIcebergCatalog, SqlServerConnectionDetails, SshConnection, SshTunnel, TlsIdentity,
+    Tunnel,
 };
 
 use crate::names::Aug;
@@ -66,6 +67,7 @@ generate_extracted_config!(
     (PublicKey1, String),
     (PublicKey2, String),
     (Region, String),
+    (Registry, String),
     (SaslMechanisms, String),
     (SaslPassword, with_options::Secret),
     (SaslUsername, StringOrSecret),
@@ -115,6 +117,7 @@ pub(super) fn validate_options_per_connection_type(
         ]
         .as_slice(),
         CreateConnectionType::AwsPrivatelink => &[AvailabilityZones, Port, ServiceName],
+        CreateConnectionType::GlueSchemaRegistry => &[AwsConnection, Registry],
         CreateConnectionType::Csr => &[
             AwsPrivatelink,
             Password,
@@ -400,6 +403,23 @@ impl ConnectionOptionExtracted {
                     tunnel,
                 })
             }
+            CreateConnectionType::GlueSchemaRegistry => {
+                scx.require_feature_flag(&vars::ENABLE_GLUE_SCHEMA_REGISTRY)?;
+
+                let aws_connection = get_aws_connection_reference(scx, &self)?
+                    .ok_or_else(|| sql_err!("AWS CONNECTION option is required"))?;
+                let registry_name = self
+                    .registry
+                    .ok_or_else(|| sql_err!("REGISTRY option is required"))?;
+                if registry_name.is_empty() {
+                    sql_bail!("invalid CONNECTION: REGISTRY must not be empty");
+                }
+
+                ConnectionDetails::GlueSchemaRegistry(GlueSchemaRegistryConnection {
+                    aws_connection,
+                    registry_name,
+                })
+            }
             CreateConnectionType::Postgres => {
                 let cert = self.ssl_certificate;
                 let key = self.ssl_key.map(|secret| secret.into());

src/sql/src/session/vars/definitions.rs

@@ -1934,6 +1934,12 @@ feature_flags!(
         default: false,
         enable_for_item_parsing: true,
     },
+    {
+        name: enable_glue_schema_registry,
+        desc: "CREATE CONNECTION ... TO AWS GLUE SCHEMA REGISTRY",
+        default: false,
+        enable_for_item_parsing: true,
+    },
     {
         name: enable_alter_set_cluster,
         desc: "ALTER ... SET CLUSTER syntax",