[dnr] SQL Server Source

src/adapter/src/catalog/builtin_table_updates.rs

@@ -1086,6 +1086,7 @@ impl CatalogState {
                     ConnectionDetails::AwsPrivatelink(..) => "aws-privatelink",
                     ConnectionDetails::Ssh { .. } => "ssh-tunnel",
                     ConnectionDetails::MySql { .. } => "mysql",
+                    ConnectionDetails::SqlServer(_) => "sql-server",
                 }),
                 Datum::String(&owner_id.to_string()),
                 privileges,
@@ -1128,7 +1129,8 @@ impl CatalogState {
             }
             ConnectionDetails::Csr(_)
             | ConnectionDetails::Postgres(_)
-            | ConnectionDetails::MySql(_) => (),
+            | ConnectionDetails::MySql(_)
+            | ConnectionDetails::SqlServer(_) => (),
         };
         updates
     }

src/adapter/src/catalog/state.rs

@@ -2637,6 +2637,7 @@ impl ConnectionResolver for CatalogState {
             Aws(conn) => Aws(conn),
             AwsPrivatelink(conn) => AwsPrivatelink(conn),
             MySql(conn) => MySql(conn.into_inline_connection(self)),
+            SqlServer(conn) => SqlServer(conn.into_inline_connection(self)),
         }
     }
 }

src/adapter/src/coord/ddl.rs

@@ -44,7 +44,7 @@ use mz_sql::session::vars::{
     MAX_CREDIT_CONSUMPTION_RATE, MAX_DATABASES, MAX_KAFKA_CONNECTIONS, MAX_MATERIALIZED_VIEWS,
     MAX_MYSQL_CONNECTIONS, MAX_NETWORK_POLICIES, MAX_OBJECTS_PER_SCHEMA, MAX_POSTGRES_CONNECTIONS,
     MAX_REPLICAS_PER_CLUSTER, MAX_ROLES, MAX_SCHEMAS_PER_DATABASE, MAX_SECRETS, MAX_SINKS,
-    MAX_SOURCES, MAX_TABLES,
+    MAX_SOURCES, MAX_SQL_SERVER_CONNECTIONS, MAX_TABLES,
 };
 use mz_storage_client::controller::{CollectionDescription, DataSource, ExportDescription};
 use mz_storage_types::connections::inline::IntoInlineConnection;
@@ -1402,6 +1402,7 @@ impl Coordinator {
         let mut new_kafka_connections = 0;
         let mut new_postgres_connections = 0;
         let mut new_mysql_connections = 0;
+        let mut new_sql_server_connections = 0;
         let mut new_aws_privatelink_connections = 0;
         let mut new_tables = 0;
         let mut new_sources = 0;
@@ -1469,6 +1470,7 @@ impl Coordinator {
                             ConnectionDetails::Kafka(_) => new_kafka_connections += 1,
                             ConnectionDetails::Postgres(_) => new_postgres_connections += 1,
                             ConnectionDetails::MySql(_) => new_mysql_connections += 1,
+                            ConnectionDetails::SqlServer(_) => new_sql_server_connections += 1,
                             ConnectionDetails::AwsPrivatelink(_) => {
                                 new_aws_privatelink_connections += 1
                             }
@@ -1639,6 +1641,7 @@ impl Coordinator {
         let mut current_aws_privatelink_connections = 0;
         let mut current_postgres_connections = 0;
         let mut current_mysql_connections = 0;
+        let mut current_sql_server_connections = 0;
         let mut current_kafka_connections = 0;
         for c in self.catalog().user_connections() {
             let connection = c
@@ -1649,6 +1652,7 @@ impl Coordinator {
                 ConnectionDetails::AwsPrivatelink(_) => current_aws_privatelink_connections += 1,
                 ConnectionDetails::Postgres(_) => current_postgres_connections += 1,
                 ConnectionDetails::MySql(_) => current_mysql_connections += 1,
+                ConnectionDetails::SqlServer(_) => current_sql_server_connections += 1,
                 ConnectionDetails::Kafka(_) => current_kafka_connections += 1,
                 ConnectionDetails::Csr(_)
                 | ConnectionDetails::Ssh { .. }
@@ -1676,6 +1680,13 @@ impl Coordinator {
             "MySQL Connection",
             MAX_MYSQL_CONNECTIONS.name(),
         )?;
+        self.validate_resource_limit(
+            current_sql_server_connections,
+            new_sql_server_connections,
+            SystemVars::max_sql_server_connections,
+            "SQL Server Connection",
+            MAX_SQL_SERVER_CONNECTIONS.name(),
+        )?;
         self.validate_resource_limit(
             current_aws_privatelink_connections,
             new_aws_privatelink_connections,

src/adapter/src/coord/sequencer/inner.rs

@@ -313,6 +313,7 @@ impl Coordinator {
                     match ingestion.desc.connection {
                         GenericSourceConnection::Postgres(_)
                         | GenericSourceConnection::MySql(_)
+                        | GenericSourceConnection::SqlServer(_)
                         | GenericSourceConnection::Kafka(_)
                         | GenericSourceConnection::LoadGenerator(_) => {
                             if let Some(cluster) = self.catalog().try_get_cluster(cluster_id) {

src/catalog/src/memory/objects.rs

@@ -1190,7 +1190,9 @@ impl Source {
                     // These multi-output sources do not use their primary
                     // source's data shard, so we don't include it in accounting
                     // for users.
-                    GenericSourceConnection::Postgres(_) | GenericSourceConnection::MySql(_) => 0,
+                    GenericSourceConnection::Postgres(_)
+                    | GenericSourceConnection::MySql(_)
+                    | GenericSourceConnection::SqlServer(_) => 0,
                     GenericSourceConnection::LoadGenerator(lg) => {
                         // TODO: make this a method on the load generator.
                         if lg.load_generator.views().is_empty() {

src/sql-server-util/BUILD.bazel

@@ -30,7 +30,11 @@ rust_library(
     rustc_env = {},
     rustc_flags = [],
     version = "0.1.0",
-    deps = [] + all_crate_deps(normal = True),
+    deps = [
+        "//src/ore:mz_ore",
+        "//src/repr:mz_repr",
+        "//src/ssh-util:mz_ssh_util",
+    ] + all_crate_deps(normal = True),
 )
 
 alias(
@@ -60,7 +64,11 @@ rust_test(
     rustc_env = {},
     rustc_flags = [],
     version = "0.1.0",
-    deps = [] + all_crate_deps(
+    deps = [
+        "//src/ore:mz_ore",
+        "//src/repr:mz_repr",
+        "//src/ssh-util:mz_ssh_util",
+    ] + all_crate_deps(
         normal = True,
         normal_dev = True,
     ),
@@ -69,7 +77,11 @@ rust_test(
 rust_doc_test(
     name = "mz_sql_server_util_doc_test",
     crate = ":mz_sql_server_util",
-    deps = [] + all_crate_deps(
+    deps = [
+        "//src/ore:mz_ore",
+        "//src/repr:mz_repr",
+        "//src/ssh-util:mz_ssh_util",
+    ] + all_crate_deps(
         normal = True,
         normal_dev = True,
     ),

src/sql-server-util/Cargo.toml

@@ -13,9 +13,16 @@ workspace = true
 anyhow = "1.0.66"
 derivative = "2.2.0"
 futures = "0.3.31"
+mz-ore = { path = "../ore", features = ["async"] }
+mz-repr = { path = "../repr" }
+mz-ssh-util = { path = "../ssh-util" }
+proptest = { version = "1.6.0", default-features = false, features = ["std"] }
+proptest-derive = { version = "0.5.1", features = ["boxed_union"] }
+serde = { version = "1.0.218", features = ["derive"] }
 smallvec = { version = "1.14.0", features = ["union"] }
 static_assertions = "1.1"
-tiberius = { version = "0.12", default-features = false, features = ["tds73"] }
+thiserror = "2.0.11"
+tiberius = { version = "0.12", features = ["rustls", "sql-browser-tokio", "tds73"], default-features = false }
 tokio = { version = "1.44.1", features = ["net"] }
 tokio-util = { version = "0.7.4", features = ["compat"] }
 tracing = "0.1.37"