feat(sqlserver): part 5 add catalog introspection

crates/dbt-adapter/src/adapter/adapter_impl.rs

@@ -28,6 +28,7 @@ use crate::metadata::postgres::PostgresMetadataAdapter;
 use crate::metadata::redshift::RedshiftMetadataAdapter;
 use crate::metadata::salesforce::SalesforceMetadataAdapter;
 use crate::metadata::snowflake::SnowflakeMetadataAdapter;
+use crate::metadata::sqlserver::SqlServerMetadataAdapter;
 use crate::metadata::{self, CatalogAndSchema, MetadataAdapter};
 use crate::query_ctx::{node_id_from_state, query_ctx_from_state};
 use crate::record_batch::{RecordBatchExt, RenamedColumn};
@@ -303,6 +304,8 @@ impl AdapterImpl {
                         Fabric => {
                             Box::new(FabricMetadataAdapter::new(engine)) as Box<dyn MetadataAdapter>
                         }
+                        SqlServer => Box::new(SqlServerMetadataAdapter::new(engine))
+                            as Box<dyn MetadataAdapter>,
                         ClickHouse => Box::new(ClickHouseMetadataAdapter::new(engine))
                             as Box<dyn MetadataAdapter>,
                         Exasol => return None,
@@ -3747,6 +3750,9 @@ impl AdapterImpl {
             Impl(Fabric, engine) => {
                 fabric::list_relations(engine.as_ref(), query_ctx, conn, db_schema, token)
             }
+            Impl(SqlServer, engine) => {
+                sqlserver::list_relations(engine.as_ref(), query_ctx, conn, db_schema, token)
+            }
             Impl(
                 adapter_type @ (Postgres | Salesforce | ClickHouse | Exasol | Starburst | Athena
                 | Trino | Datafusion | Dremio | Oracle),

crates/dbt-adapter/src/metadata/get_relation.rs

@@ -76,6 +76,9 @@ pub fn get_relation(
         AdapterType::Fabric => fabric_get_relation(
             adapter, state, ctx, conn, database, schema, identifier, token,
         ),
+        AdapterType::SqlServer => sqlserver_get_relation(
+            adapter, state, ctx, conn, database, schema, identifier, token,
+        ),
         AdapterType::ClickHouse => clickhouse_get_relation(
             adapter, state, ctx, conn, database, schema, identifier, token,
         ),
@@ -1002,6 +1005,53 @@ fn fabric_get_relation(
     ))))
 }
 
+#[allow(clippy::too_many_arguments)]
+fn sqlserver_get_relation(
+    adapter: &AdapterImpl,
+    state: &State,
+    ctx: &QueryCtx,
+    conn: &'_ mut dyn Connection,
+    database: &str,
+    schema: &str,
+    identifier: &str,
+    token: CancellationToken,
+) -> AdapterResult<Option<Box<dyn BaseRelation>>> {
+    use crate::metadata::sqlserver::{build_get_relation_sql, relation_type_from_table_type};
+
+    let relation = Relation::new_sqlserver(
+        Some(database.to_string()),
+        Some(schema.to_string()),
+        Some(identifier.to_string()),
+        None,
+        adapter.quoting(),
+    );
+
+    let sql = build_get_relation_sql(&relation.quoted(database), schema, identifier);
+
+    let batch = adapter
+        .engine()
+        .execute(Some(state), conn, ctx, &sql, token)?;
+
+    if batch.num_rows() == 0 {
+        return Ok(None);
+    }
+
+    let table_types = batch.column_values::<StringArray>("table_type")?;
+    if table_types.len() != 1 {
+        return Err(AdapterError::new(
+            AdapterErrorKind::UnexpectedResult,
+            format!(
+                "Expected exactly one row for SQL Server get_relation, got {}",
+                table_types.len()
+            ),
+        ));
+    }
+
+    let relation_type = relation_type_from_table_type(table_types.value(0));
+
+    Ok(Some(Box::new(relation.with_relation_type(relation_type))))
+}
+
 #[allow(clippy::too_many_arguments)]
 fn clickhouse_get_relation(
     adapter: &AdapterImpl,

crates/dbt-adapter/src/metadata/mod.rs

@@ -27,6 +27,7 @@ pub(crate) mod redshift;
 pub(crate) mod salesforce;
 pub mod snowflake; // XXX: temporarily pub before the refactor is complete
 pub(crate) mod spark;
+pub(crate) mod sqlserver;
 pub(crate) mod view_definition;
 
 // Re-export `metadata_adapter` symbols