start, MVP of SQL Server Source purification

* implement purification for SQL Server sources, lists tables from upstream that have CDC enabled and their capture instances
* returns an error for CREATE TABLE ... FROM SOURCE for SQL Server

Author: ParkMyCar

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

@@ -1185,15 +1185,15 @@ impl_display_t!(MySqlConfigOption);
 
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub enum SqlServerConfigOptionName {
-    /// The name for the "capture job" that will get spawn in SQL Server to
-    /// populate the change table that we read from.
-    CaptureInstance,
+    /// Hex encoded string of binary serialization of
+    /// `mz_storage_types::sources::sql_server::SqlServerSourceDetails`.
+    Details,
 }
 
 impl AstDisplay for SqlServerConfigOptionName {
     fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
         f.write_str(match self {
-            SqlServerConfigOptionName::CaptureInstance => "CAPTURE INSTANCE",
+            SqlServerConfigOptionName::Details => "DETAILS",
         })
     }
 }
@@ -1207,7 +1207,7 @@ impl WithOptionName for SqlServerConfigOptionName {
     /// on the conservative side and return `true`.
     fn redact_value(&self) -> bool {
         match self {
-            SqlServerConfigOptionName::CaptureInstance => false,
+            SqlServerConfigOptionName::Details => false,
         }
     }
 }

src/sql-parser/src/parser.rs

@@ -3469,11 +3469,8 @@ impl<'a> Parser<'a> {
     fn parse_sql_server_connection_option(
         &mut self,
     ) -> Result<SqlServerConfigOption<Raw>, ParserError> {
-        let name = match self.expect_one_of_keywords(&[CAPTURE])? {
-            CAPTURE => {
-                self.expect_keyword(INSTANCE)?;
-                SqlServerConfigOptionName::CaptureInstance
-            }
+        let name = match self.expect_one_of_keywords(&[DETAILS])? {
+            DETAILS => SqlServerConfigOptionName::Details,
             _ => unreachable!(),
         };
 

src/sql-server-util/src/desc.rs

@@ -114,6 +114,8 @@ pub struct SqlServerTableRaw {
     pub schema_name: Arc<str>,
     /// Name of the table.
     pub name: Arc<str>,
+    /// The capture instance replicating changes.
+    pub capture_instance: Arc<str>,
     /// Columns for the table.
     pub columns: Arc<[SqlServerColumnRaw]>,
     /// Whether or not CDC is enabled for this table.
@@ -543,6 +545,7 @@ impl RustType<proto_sql_server_column_desc::DecodeType> for SqlServerColumnDecod
 ///
 /// The goal of this type is to perform any expensive "downcasts" so in the hot
 /// path of decoding rows we do the minimal amount of work.
+#[derive(Debug)]
 pub struct SqlServerRowDecoder {
     decoders: Vec<(Arc<str>, ColumnType, SqlServerColumnDecodeType)>,
 }
@@ -701,6 +704,7 @@ mod tests {
         let sql_server_desc = SqlServerTableRaw {
             schema_name: "my_schema".into(),
             name: "my_table".into(),
+            capture_instance: "my_table_CT".into(),
             columns: sql_server_columns.into(),
             is_cdc_enabled: true,
         };

src/sql-server-util/src/inspect.rs

@@ -11,9 +11,12 @@
 
 use itertools::Itertools;
 use smallvec::SmallVec;
+use std::collections::BTreeMap;
+use std::fmt;
 use std::sync::Arc;
 
 use crate::cdc::{Lsn, RowFilterOption};
+use crate::desc::{SqlServerColumnRaw, SqlServerTableRaw};
 use crate::{Client, SqlServerError};
 
 /// Returns the minimum log sequence number for the specified `capture_instance`.
@@ -159,3 +162,151 @@ WHERE c.capture_instance IN ({param_indexes});"
 
     Ok(tables)
 }
+
+/// Ensure change data capture (CDC) is enabled for the database the provided
+/// `client` is currently connected to.
+///
+/// See: <https://learn.microsoft.com/en-us/sql/relational-databases/track-changes/enable-and-disable-change-data-capture-sql-server?view=sql-server-ver16>
+pub async fn ensure_database_cdc_enabled(client: &mut Client) -> Result<(), SqlServerError> {
+    static DATABASE_CDC_ENABLED_QUERY: &str =
+        "SELECT is_cdc_enabled FROM sys.databases WHERE database_id = DB_ID();";
+    let result = client.simple_query(DATABASE_CDC_ENABLED_QUERY).await?;
+
+    check_system_result(&result, "database CDC".to_string(), true)?;
+    Ok(())
+}
+
+/// Ensure change data capture (CDC) is enabled for the specified table.
+///
+/// See: <https://learn.microsoft.com/en-us/sql/relational-databases/track-changes/enable-and-disable-change-data-capture-sql-server?view=sql-server-ver16#enable-for-a-table>
+pub async fn ensure_table_cdc_enabled(
+    client: &mut Client,
+    schema: &str,
+    table: &str,
+) -> Result<(), SqlServerError> {
+    static TABLE_CDC_ENABLED_QUERY: &str = "
+SELECT is_tracked_by_cdc FROM sys.tables tables
+JOIN sys.schemas schemas
+ON tables.schema_id = schemas.schema_id
+WHERE schemas.name = @P1 AND tables.name = @P2;
+";
+    let result = client
+        .query(TABLE_CDC_ENABLED_QUERY, &[&schema, &table])
+        .await?;
+
+    check_system_result(&result, "table CDC".to_string(), true)?;
+    Ok(())
+}
+
+/// Ensure the `SNAPSHOT` transaction isolation level is enabled for the
+/// database the provided `client` is currently connected to.
+///
+/// See: <https://learn.microsoft.com/en-us/sql/t-sql/statements/set-transaction-isolation-level-transact-sql?view=sql-server-ver16>
+pub async fn ensure_snapshot_isolation_enabled(client: &mut Client) -> Result<(), SqlServerError> {
+    static SNAPSHOT_ISOLATION_QUERY: &str =
+        "SELECT snapshot_isolation_state FROM sys.databases WHERE database_id = DB_ID();";
+    let result = client.simple_query(SNAPSHOT_ISOLATION_QUERY).await?;
+
+    check_system_result(&result, "snapshot isolation".to_string(), 1u8)?;
+    Ok(())
+}
+
+pub async fn get_tables(client: &mut Client) -> Result<Vec<SqlServerTableRaw>, SqlServerError> {
+    static GET_TABLES_QUERY: &str = "
+SELECT
+    s.name as schema_name,
+    t.name as table_name,
+    ch.capture_instance as capture_instance,
+    c.name as col_name,
+    ty.name as col_type,
+    c.is_nullable as col_nullable,
+    c.max_length as col_max_length,
+    c.precision as col_precision,
+    c.scale as col_scale
+FROM sys.tables t
+JOIN sys.schemas s ON t.schema_id = s.schema_id
+JOIN sys.columns c ON t.object_id = c.object_id
+JOIN sys.types ty ON c.system_type_id = ty.system_type_id
+JOIN cdc.change_tables ch ON t.object_id = ch.source_object_id
+";
+    fn get_value<'a, T: tiberius::FromSql<'a>>(
+        row: &'a tiberius::Row,
+        name: &'static str,
+    ) -> Result<T, SqlServerError> {
+        row.try_get(name)?
+            .ok_or(SqlServerError::MissingColumn(name))
+    }
+
+    let result = client.simple_query(GET_TABLES_QUERY).await?;
+
+    // Group our columns by (schema, name).
+    let mut tables = BTreeMap::default();
+    for row in result {
+        let schema_name: Arc<str> = get_value::<&str>(&row, "schema_name")?.into();
+        let table_name: Arc<str> = get_value::<&str>(&row, "table_name")?.into();
+        let capture_instance: Arc<str> = get_value::<&str>(&row, "capture_instance")?.into();
+
+        let column_name = get_value::<&str>(&row, "col_name")?.into();
+        let column = SqlServerColumnRaw {
+            name: Arc::clone(&column_name),
+            data_type: get_value::<&str>(&row, "col_type")?.into(),
+            is_nullable: get_value(&row, "col_nullable")?,
+            max_length: get_value(&row, "col_max_length")?,
+            precision: get_value(&row, "col_precision")?,
+            scale: get_value(&row, "col_scale")?,
+        };
+
+        let columns = tables
+            .entry((
+                Arc::clone(&schema_name),
+                Arc::clone(&table_name),
+                Arc::clone(&capture_instance),
+            ))
+            .or_insert_with(|| Vec::default());
+        columns.push(column);
+    }
+
+    // Flatten into our raw Table description.
+    let tables = tables
+        .into_iter()
+        .map(|((schema, name, capture_instance), columns)| {
+            Ok::<_, SqlServerError>(SqlServerTableRaw {
+                schema_name: schema,
+                name,
+                capture_instance,
+                columns: columns.into(),
+                is_cdc_enabled: true,
+            })
+        })
+        .collect::<Result<_, _>>()?;
+
+    Ok(tables)
+}
+
+/// Helper function to parse an expected result from a "system" query.
+fn check_system_result<'a, T>(
+    result: &'a SmallVec<[tiberius::Row; 1]>,
+    name: String,
+    expected: T,
+) -> Result<(), SqlServerError>
+where
+    T: tiberius::FromSql<'a> + Copy + fmt::Debug + fmt::Display + PartialEq,
+{
+    match &result[..] {
+        [row] => {
+            let result: Option<T> = row.try_get(0)?;
+            if result == Some(expected) {
+                Ok(())
+            } else {
+                Err(SqlServerError::InvalidSystemSetting {
+                    name,
+                    expected: expected.to_string(),
+                    actual: format!("{result:?}"),
+                })
+            }
+        }
+        other => Err(SqlServerError::InvariantViolated(format!(
+            "expected 1 row, got {other:?}"
+        ))),
+    }
+}

src/sql-server-util/src/lib.rs

@@ -739,6 +739,8 @@ pub enum SqlServerError {
     SqlServer(#[from] tiberius::error::Error),
     #[error(transparent)]
     CdcError(#[from] crate::cdc::CdcError),
+    #[error("expected column '{0}' to be present")]
+    MissingColumn(&'static str),
     #[error("'{column_type}' from column '{column_name}' is not supported: {reason}")]
     UnsupportedDataType {
         column_name: String,
@@ -749,6 +751,12 @@ pub enum SqlServerError {
     IO(#[from] tokio::io::Error),
     #[error("found invalid data in the column '{column_name}': {error}")]
     InvalidData { column_name: String, error: String },
+    #[error("invalid SQL Server system setting '{name}'. Expected '{expected}'. Got '{actual}'.")]
+    InvalidSystemSetting {
+        name: String,
+        expected: String,
+        actual: String,
+    },
     #[error("invariant was violated: {0}")]
     InvariantViolated(String),
     #[error(transparent)]

src/sql/src/plan/error.rs

@@ -44,6 +44,7 @@ use crate::plan::typeconv::CastContext;
 use crate::pure::error::{
     CsrPurificationError, KafkaSinkPurificationError, KafkaSourcePurificationError,
     LoadGeneratorSourcePurificationError, MySqlSourcePurificationError, PgSourcePurificationError,
+    SqlServerSourcePurificationError,
 };
 use crate::session::vars::VarError;
 
@@ -250,6 +251,7 @@ pub enum PlanError {
     LoadGeneratorSourcePurification(LoadGeneratorSourcePurificationError),
     CsrPurification(CsrPurificationError),
     MySqlSourcePurification(MySqlSourcePurificationError),
+    SqlServerSourcePurificationError(SqlServerSourcePurificationError),
     UseTablesForSources(String),
     MissingName(CatalogItemType),
     InvalidRefreshAt,
@@ -331,6 +333,7 @@ impl PlanError {
             Self::InternalFunctionCall => Some("This function is for the internal use of the database system and cannot be called directly.".into()),
             Self::PgSourcePurification(e) => e.detail(),
             Self::MySqlSourcePurification(e) => e.detail(),
+            Self::SqlServerSourcePurificationError(e) => e.detail(),
             Self::KafkaSourcePurification(e) => e.detail(),
             Self::LoadGeneratorSourcePurification(e) => e.detail(),
             Self::CsrPurification(e) => e.detail(),
@@ -439,6 +442,8 @@ impl PlanError {
             Self::Catalog(e) => e.hint(),
             Self::VarError(e) => e.hint(),
             Self::PgSourcePurification(e) => e.hint(),
+            Self::MySqlSourcePurification(e) => e.hint(),
+            Self::SqlServerSourcePurificationError(e) => e.hint(),
             Self::KafkaSourcePurification(e) => e.hint(),
             Self::LoadGeneratorSourcePurification(e) => e.hint(),
             Self::CsrPurification(e) => e.hint(),
@@ -739,6 +744,7 @@ impl fmt::Display for PlanError {
             Self::KafkaSinkPurification(e) => write!(f, "KAFKA sink validation: {}", e),
             Self::CsrPurification(e) => write!(f, "CONFLUENT SCHEMA REGISTRY validation: {}", e),
             Self::MySqlSourcePurification(e) => write!(f, "MYSQL source validation: {}", e),
+            Self::SqlServerSourcePurificationError(e) => write!(f, "SQL SERVER source validation: {}", e),
             Self::UseTablesForSources(command) => write!(f, "{command} not supported; use CREATE TABLE .. FROM SOURCE instead"),
             Self::MangedReplicaName(name) => {
                 write!(f, "{name} is reserved for replicas of managed clusters")
@@ -940,6 +946,12 @@ impl From<MySqlSourcePurificationError> for PlanError {
     }
 }
 
+impl From<SqlServerSourcePurificationError> for PlanError {
+    fn from(e: SqlServerSourcePurificationError) -> Self {
+        PlanError::SqlServerSourcePurificationError(e)
+    }
+}
+
 impl From<IdentError> for PlanError {
     fn from(e: IdentError) -> Self {
         PlanError::InvalidIdent(e)

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

@@ -532,7 +532,7 @@ generate_extracted_config!(
     (ExcludeColumns, Vec::<UnresolvedItemName>, Default(vec![]))
 );
 
-generate_extracted_config!(SqlServerConfigOption, (CaptureInstance, String));
+generate_extracted_config!(SqlServerConfigOption, (Details, String));
 
 pub fn plan_create_webhook_source(
     scx: &StatementContext,