use std::collections::{BTreeMap, HashMap, HashSet};
use async_trait::async_trait;
use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime, NaiveTime, Utc};
use futures::TryStreamExt;
use rustium_column_transform::ColumnTransformer;
use rustium_config::{SnapshotConfig, SnapshotMode, SqlServerSourceConfig};
use rustium_core::{
ChangeEvent, DataValue, Error, EventId, EventSchema, FieldSchema, Operation, RecordBoundary,
Result, RetryPolicy, Row, SignalAcknowledgement, SignalRecord, SourceConnector, SourceContext,
SourceMetadata, SourcePosition, SourceRecord, SqlServerPosition, TransactionMetadata,
};
use tiberius::{
AuthMethod, Client, ColumnData, Config as TdsConfig, EncryptionLevel, Row as TdsRow,
};
use tokio::net::TcpStream;
use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
use tracing::{info, warn};
use crate::state::{
IncrementalSnapshotProgress, SqlServerKeyValue, decode_connector_state, encode_connector_state,
};
const CONNECTOR_VERSION: &str = env!("CARGO_PKG_VERSION");
const LSN_SIZE: usize = 10;
const RAW_DELETE: i32 = 1;
const RAW_INSERT: i32 = 2;
const RAW_UPDATE_BEFORE: i32 = 3;
const RAW_UPDATE_AFTER: i32 = 4;
const COMMIT_SERIAL: u64 = 5;
const MAX_COMPLETED_SIGNAL_IDS: usize = 1_024;
const WINDOW_OPEN_SIGNAL: &str = "snapshot-window-open";
const WINDOW_CLOSE_SIGNAL: &str = "snapshot-window-close";
type SqlClient = Client<Compat<TcpStream>>;
#[derive(Debug, Clone)]
struct CaptureTable {
schema: String,
table: String,
capture_instance: String,
source_object_id: i32,
event_schema: EventSchema,
}
impl CaptureTable {
fn key(&self) -> (String, String) {
(self.schema.clone(), self.table.clone())
}
}
#[derive(Debug, Clone)]
struct CdcCursor {
commit_lsn: Vec<u8>,
change_lsn: Vec<u8>,
raw_operation: i32,
}
impl CdcCursor {
fn at_snapshot(commit_lsn: Vec<u8>) -> Self {
Self {
commit_lsn,
change_lsn: max_lsn_bytes(),
raw_operation: i32::try_from(COMMIT_SERIAL).unwrap_or(i32::MAX),
}
}
fn from_position(
position: &SqlServerPosition,
connector_state_boundary: bool,
) -> Result<(Self, Option<SourcePosition>)> {
if position.snapshot || position.event_serial == COMMIT_SERIAL || connector_state_boundary {
return Ok((Self::at_snapshot(parse_lsn(&position.commit_lsn)?), None));
}
Ok((
Self {
commit_lsn: parse_lsn(&position.commit_lsn)?,
change_lsn: zero_lsn_bytes(),
raw_operation: 0,
},
Some(SourcePosition::SqlServer(position.clone())),
))
}
fn commit_complete(&self) -> bool {
self.raw_operation == i32::try_from(COMMIT_SERIAL).unwrap_or(i32::MAX)
&& self.change_lsn == max_lsn_bytes()
}
}
#[derive(Debug)]
struct RawChange {
commit_lsn: Vec<u8>,
change_lsn: Vec<u8>,
raw_operation: i32,
capture_instance: String,
row: Row,
source_time: Option<DateTime<Utc>>,
}
#[derive(Debug)]
struct ActiveTransaction {
commit_lsn: Vec<u8>,
source_time: Option<DateTime<Utc>>,
total_order: u64,
collection_order: HashMap<(String, String), u64>,
}
#[derive(Debug, Clone)]
enum SnapshotSignal {
Execute {
id: String,
data_collections: Vec<String>,
additional_conditions: BTreeMap<String, String>,
},
Stop {
id: String,
data_collections: Vec<String>,
},
Pause {
id: String,
},
Resume {
id: String,
},
Unsupported {
id: String,
signal_type: String,
},
}
struct SqlServerIncrementalSnapshot {
progress: Option<IncrementalSnapshotProgress>,
opening_window_id: Option<String>,
window: Option<SqlServerIncrementalWindow>,
completed_signal_ids: Vec<String>,
event_serial: u64,
state_dirty: bool,
}
impl SqlServerIncrementalSnapshot {
fn new(
progress: Option<IncrementalSnapshotProgress>,
completed_signal_ids: Vec<String>,
) -> Self {
Self {
progress,
opening_window_id: None,
window: None,
completed_signal_ids,
event_serial: 0,
state_dirty: false,
}
}
fn progress(&self) -> Option<&IncrementalSnapshotProgress> {
self.progress.as_ref()
}
fn completed_signal_ids(&self) -> &[String] {
&self.completed_signal_ids
}
fn is_active(&self) -> bool {
self.progress
.as_ref()
.is_some_and(|progress| !progress.paused)
&& self.opening_window_id.is_none()
&& self.window.is_none()
}
fn discard_window(&mut self) {
self.opening_window_id = None;
self.window = None;
}
fn state_dirty(&self) -> bool {
self.state_dirty
}
fn mark_checkpointed(&mut self) {
self.state_dirty = false;
}
fn remember_completed(&mut self, id: String) {
if let Some(index) = self
.completed_signal_ids
.iter()
.position(|candidate| candidate == &id)
{
self.completed_signal_ids.remove(index);
}
self.completed_signal_ids.push(id);
if self.completed_signal_ids.len() > MAX_COMPLETED_SIGNAL_IDS {
self.completed_signal_ids.remove(0);
}
}
fn parse_external_record(record: &SignalRecord) -> Result<SnapshotSignal> {
if record.id.trim().is_empty() || record.signal_type.trim().is_empty() {
return Err(Error::Source(
"SQL Server external signal requires non-empty id and type".into(),
));
}
parse_snapshot_signal(&record.id, &record.signal_type, &record.data)
}
fn parse_row(row: &Row) -> Result<SnapshotSignal> {
let id = signal_text(row, "id")?;
let signal_type = signal_text(row, "type")?;
let data = signal_text(row, "data")?;
let value = serde_json::from_str::<serde_json::Value>(&data).map_err(|error| {
Error::Source(format!(
"SQL Server signal {id:?} has invalid JSON data: {error}"
))
})?;
parse_snapshot_signal(&id, &signal_type, &value)
}
fn handle_signal(&mut self, signal: SnapshotSignal, source: &SqlServerSource) -> Result<()> {
match signal {
SnapshotSignal::Execute {
id,
data_collections,
additional_conditions,
} => {
if self.completed_signal_ids.contains(&id)
|| self
.progress
.as_ref()
.is_some_and(|progress| progress.signal_id == id)
{
return Ok(());
}
if self.progress.is_some() {
tracing::warn!(%id, "SQL Server incremental snapshot is already active; execute signal ignored");
return Ok(());
}
source.signal_capture().ok_or_else(|| {
Error::Configuration(
"SQL Server incremental snapshots require a CDC-enabled signal.data.collection so Rustium can emit open/close watermarks"
.into(),
)
})?;
self.progress = Some(IncrementalSnapshotProgress {
signal_id: id,
data_collections: source.expand_data_collections(&data_collections)?,
additional_conditions,
current_collection: 0,
last_key: None,
maximum_key: None,
chunk_sequence: 1,
paused: false,
});
self.state_dirty = true;
Ok(())
}
SnapshotSignal::Stop {
id,
data_collections,
} => {
let Some(mut progress) = self.progress.take() else {
return Ok(());
};
if data_collections.is_empty() {
self.discard_window();
self.remember_completed(progress.signal_id);
self.state_dirty = true;
info!(%id, "SQL Server incremental snapshot stopped");
return Ok(());
}
let patterns = compile_collection_patterns(&data_collections)?;
let original = progress.data_collections.clone();
let current = original.get(progress.current_collection).cloned();
let retained_before = original
.iter()
.take(progress.current_collection)
.filter(|collection| !collection_matches_any(collection, &patterns))
.count();
progress
.data_collections
.retain(|collection| !collection_matches_any(collection, &patterns));
if progress.data_collections.len() == original.len() {
self.progress = Some(progress);
return Ok(());
}
self.discard_window();
if current
.as_ref()
.is_some_and(|collection| collection_matches_any(collection, &patterns))
{
progress.last_key = None;
progress.maximum_key = None;
}
progress.current_collection = retained_before;
if progress.current_collection >= progress.data_collections.len() {
self.remember_completed(progress.signal_id);
} else {
self.progress = Some(progress);
}
self.state_dirty = true;
info!(%id, "SQL Server incremental snapshot collections stopped");
Ok(())
}
SnapshotSignal::Pause { id } => {
if let Some(progress) = &mut self.progress
&& !progress.paused
{
progress.paused = true;
self.state_dirty = true;
info!(%id, "SQL Server incremental snapshot paused");
}
Ok(())
}
SnapshotSignal::Resume { id } => {
if let Some(progress) = &mut self.progress
&& progress.paused
{
progress.paused = false;
self.state_dirty = true;
info!(%id, "SQL Server incremental snapshot resumed");
}
Ok(())
}
SnapshotSignal::Unsupported { id, signal_type } => {
tracing::warn!(%id, %signal_type, "unsupported SQL Server runtime signal ignored");
Ok(())
}
}
}
async fn start_next_chunk(&mut self, source: &SqlServerSource) -> Result<()> {
if self.opening_window_id.is_some() || self.window.is_some() {
return Ok(());
}
let Some(progress) = self.progress.clone() else {
return Ok(());
};
if progress.paused {
return Ok(());
}
let window_id = uuid::Uuid::new_v4().to_string();
source
.emit_incremental_watermark(&format!("{window_id}-open"), WINDOW_OPEN_SIGNAL)
.await?;
self.opening_window_id = Some(window_id);
Ok(())
}
async fn open_window(&mut self, source: &SqlServerSource, window_id: String) -> Result<()> {
let Some(progress) = self.progress.clone() else {
return Ok(());
};
let collection = progress
.data_collections
.get(progress.current_collection)
.cloned()
.ok_or_else(|| {
Error::State(format!(
"SQL Server incremental snapshot collection index {} is outside {} collections",
progress.current_collection,
progress.data_collections.len()
))
})?;
let capture = source.capture_for_collection(&collection).ok_or_else(|| {
Error::Source(format!(
"SQL Server incremental snapshot collection {collection:?} is not captured"
))
})?;
let chunk = source.read_incremental_chunk(&capture, &progress).await?;
let row_count = chunk.rows.len();
let last_key = chunk.rows.last().map(|(_, key)| key.clone());
let remaining_keys = chunk
.rows
.iter()
.map(|(_, key)| key.clone())
.collect::<HashSet<_>>();
self.window = Some(SqlServerIncrementalWindow {
id: window_id.clone(),
collection,
capture,
rows: chunk.rows,
remaining_keys,
maximum_key: chunk.maximum_key,
last_key,
row_count,
close_commit_lsn: None,
});
source
.emit_incremental_watermark(&format!("{window_id}-close"), WINDOW_CLOSE_SIGNAL)
.await?;
Ok(())
}
async fn handle_internal_signal(
&mut self,
record: &SourceRecord,
row: &Row,
source: &SqlServerSource,
) -> Result<bool> {
let signal_type = signal_text(row, "type")?;
if !matches!(
signal_type.as_str(),
WINDOW_OPEN_SIGNAL | WINDOW_CLOSE_SIGNAL
) {
return Ok(false);
}
let id = signal_text(row, "id")?;
let opening_matches = self
.opening_window_id
.as_ref()
.is_some_and(|window_id| id == format!("{window_id}-open"));
let closing_matches = self
.window
.as_ref()
.is_some_and(|window| id == format!("{}-close", window.id));
if signal_type == WINDOW_OPEN_SIGNAL && opening_matches {
let window_id = self.opening_window_id.take().ok_or_else(|| {
Error::State("SQL Server incremental opening window disappeared".into())
})?;
self.open_window(source, window_id).await?;
} else if signal_type == WINDOW_CLOSE_SIGNAL && closing_matches {
let SourcePosition::SqlServer(position) = &record.position else {
return Err(Error::State(
"SQL Server watermark record has a non-SQL Server position".into(),
));
};
let capture = self
.window
.as_ref()
.map(|window| window.capture.clone())
.ok_or_else(|| Error::State("SQL Server incremental window disappeared".into()))?;
source.validate_incremental_schema(&capture).await?;
if let Some(window) = &mut self.window {
window.close_commit_lsn = Some(position.commit_lsn.clone());
}
}
Ok(true)
}
fn observe_record(&mut self, record: &SourceRecord) -> Result<()> {
let Some(window) = &mut self.window else {
return Ok(());
};
if window.close_commit_lsn.is_some() {
return Ok(());
}
let Some(event) = &record.event else {
return Ok(());
};
let collection = event
.source
.schema
.as_deref()
.zip(event.source.table.as_deref())
.map(|(schema, table)| format!("{schema}.{table}"));
if collection.as_deref() != Some(window.collection.as_str()) {
return Ok(());
}
for row in [event.before.as_ref(), event.after.as_ref()]
.into_iter()
.flatten()
{
let key = sqlserver_key_from_row(row, &window.capture.event_schema)?;
window.remaining_keys.remove(&key);
}
Ok(())
}
fn closes_at(&self, record: &SourceRecord) -> bool {
let SourcePosition::SqlServer(position) = &record.position else {
return false;
};
record.boundary == RecordBoundary::TransactionCommit
&& self.window.as_ref().is_some_and(|window| {
window.close_commit_lsn.as_deref() == Some(position.commit_lsn.as_str())
})
}
async fn finish_window(
&mut self,
source: &SqlServerSource,
base_position: &SourcePosition,
output: &tokio::sync::mpsc::Sender<Result<SourceRecord>>,
) -> Result<SourcePosition> {
let window = self.window.take().ok_or_else(|| {
Error::State("SQL Server incremental snapshot has no pending window".into())
})?;
let progress = self.progress.clone().ok_or_else(|| {
Error::State("SQL Server incremental snapshot window has no progress".into())
})?;
for (row, key) in window.rows {
if !window.remaining_keys.contains(&key) {
continue;
}
self.event_serial = self.event_serial.saturating_add(1);
let position = sqlserver_incremental_position(base_position, self.event_serial)?;
let mut attributes = BTreeMap::new();
attributes.insert("rustium.snapshot.kind".into(), "incremental".into());
let event = ChangeEvent {
id: EventId::deterministic(
&source.connector_name,
source.database(),
&position,
&window.collection,
self.event_serial,
),
source: SourceMetadata {
connector: "sqlserver".into(),
connector_name: source.connector_name.clone(),
database: source.database().into(),
schema: Some(window.capture.schema.clone()),
table: Some(window.capture.table.clone()),
snapshot: true,
version: CONNECTOR_VERSION.into(),
attributes,
},
position: position.clone(),
transaction: None,
operation: Operation::Read,
before: None,
after: Some(row),
schema: window.capture.event_schema.clone(),
source_time: None,
observed_time: Utc::now(),
};
output
.send(Ok(SourceRecord::data(event)))
.await
.map_err(|_| Error::Cancelled)?;
}
let mut next = progress;
if next.maximum_key.is_none() {
next.maximum_key.clone_from(&window.maximum_key);
}
next.last_key = window.last_key;
let collection_complete = next.maximum_key.is_none()
|| next.last_key.as_ref() == next.maximum_key.as_ref()
|| window.row_count < source.config.incremental_snapshot_chunk_size;
if collection_complete {
next.current_collection = next.current_collection.saturating_add(1);
next.last_key = None;
next.maximum_key = None;
}
next.chunk_sequence = next.chunk_sequence.saturating_add(1);
if next.current_collection >= next.data_collections.len() {
self.remember_completed(next.signal_id);
self.progress = None;
} else {
self.progress = Some(next);
}
self.event_serial = self.event_serial.saturating_add(1);
let position = sqlserver_incremental_position(base_position, self.event_serial)?;
output
.send(Ok(SourceRecord {
event: None,
position: position.clone(),
boundary: RecordBoundary::TransactionCommit,
connector_state: Some(encode_connector_state(
self.progress.as_ref(),
&self.completed_signal_ids,
)?),
signal_acknowledgements: Vec::new(),
}))
.await
.map_err(|_| Error::Cancelled)?;
self.mark_checkpointed();
Ok(position)
}
}
struct IncrementalChunk {
rows: Vec<(Row, Vec<SqlServerKeyValue>)>,
maximum_key: Option<Vec<SqlServerKeyValue>>,
}
struct SqlServerIncrementalWindow {
id: String,
collection: String,
capture: CaptureTable,
rows: Vec<(Row, Vec<SqlServerKeyValue>)>,
remaining_keys: HashSet<Vec<SqlServerKeyValue>>,
maximum_key: Option<Vec<SqlServerKeyValue>>,
last_key: Option<Vec<SqlServerKeyValue>>,
row_count: usize,
close_commit_lsn: Option<String>,
}
pub struct SqlServerSource {
connector_name: String,
config: SqlServerSourceConfig,
snapshot: SnapshotConfig,
captures: Vec<CaptureTable>,
retry_policy: RetryPolicy,
column_transformer: Option<ColumnTransformer>,
}
impl SqlServerSource {
#[must_use]
pub fn new(
connector_name: impl Into<String>,
config: SqlServerSourceConfig,
snapshot: SnapshotConfig,
) -> Self {
Self {
connector_name: connector_name.into(),
config,
snapshot,
captures: Vec::new(),
retry_policy: RetryPolicy::default(),
column_transformer: None,
}
}
#[must_use]
pub fn with_retry_policy(mut self, retry_policy: RetryPolicy) -> Self {
self.retry_policy = retry_policy;
self
}
fn database(&self) -> &str {
&self.config.databases[0]
}
async fn validate_source(&mut self) -> Result<()> {
self.column_transformer =
Some(ColumnTransformer::new(&self.config.column_transformations)?);
let mut client = connect(&self.config, self.database()).await?;
let row = client
.simple_query(
"SELECT CAST(SERVERPROPERTY('ProductMajorVersion') AS int) AS major_version, \
CAST(is_cdc_enabled AS bit) AS is_cdc_enabled, \
CAST(snapshot_isolation_state AS int) AS snapshot_isolation_state \
FROM sys.databases WHERE name = DB_NAME()",
)
.await
.map_err(sqlserver_error)?
.into_row()
.await
.map_err(sqlserver_error)?
.ok_or_else(|| Error::Source("SQL Server did not return database metadata".into()))?;
let major_version = required::<i32>(&row, "major_version")?;
let cdc_enabled = required::<bool>(&row, "is_cdc_enabled")?;
let snapshot_isolation_state = required::<i32>(&row, "snapshot_isolation_state")?;
if major_version < 14 {
return Err(Error::Configuration(format!(
"SQL Server 2017 or newer is required; major version is {major_version}"
)));
}
if !cdc_enabled {
return Err(Error::Configuration(format!(
"CDC is not enabled for SQL Server database {:?}",
self.database()
)));
}
if self.config.snapshot_isolation_mode == "snapshot" && snapshot_isolation_state != 1 {
return Err(Error::Configuration(
"snapshot.isolation.mode=snapshot requires ALLOW_SNAPSHOT_ISOLATION ON".into(),
));
}
let captures = discover_captures(&mut client, &self.config, &self.connector_name).await?;
let signal_table = signal_table_key(&self.config);
if captures
.iter()
.all(|capture| signal_table.as_ref() == Some(&capture.key()))
{
return Err(Error::Configuration(
"the SQL Server CDC capture instances and table filters select no tables".into(),
));
}
if let Some(signal_table) = signal_table {
let capture = captures
.iter()
.find(|capture| capture.key() == signal_table)
.ok_or_else(|| {
Error::Configuration(format!(
"SQL Server signal table {}.{} is not CDC-enabled",
signal_table.0, signal_table.1
))
})?;
validate_signal_schema(capture)?;
validate_signal_insert_permission(&mut client, capture).await?;
}
current_max_lsn(&mut client).await?;
client.close().await.map_err(sqlserver_error)?;
self.captures = captures;
Ok(())
}
async fn run_snapshot(
&self,
output: &tokio::sync::mpsc::Sender<Result<SourceRecord>>,
) -> Result<Vec<u8>> {
let mut client = connect(&self.config, self.database()).await?;
client
.simple_query(snapshot_begin_sql(&self.config.snapshot_isolation_mode))
.await
.map_err(sqlserver_error)?
.into_results()
.await
.map_err(sqlserver_error)?;
let anchor = current_max_lsn(&mut client).await?;
let mut ordinal = 0_u64;
let mut captures = self.captures.clone();
let signal_table = signal_table_key(&self.config);
captures.retain(|capture| {
signal_table.as_ref() != Some(&capture.key())
&& snapshot_includes(
&self.snapshot,
self.database(),
&capture.schema,
&capture.table,
)
});
captures.sort_by_key(CaptureTable::key);
for capture in &captures {
snapshot_table(
&mut client,
self.database(),
&self.connector_name,
capture,
&anchor,
&mut ordinal,
output,
self.column_transformer
.as_ref()
.expect("validated transformer"),
)
.await?;
}
client
.simple_query("COMMIT TRANSACTION")
.await
.map_err(sqlserver_error)?
.into_results()
.await
.map_err(sqlserver_error)?;
ordinal += 1;
output
.send(Ok(SourceRecord {
event: None,
position: sqlserver_position(
self.database(),
&anchor,
&zero_lsn_bytes(),
ordinal,
true,
),
boundary: RecordBoundary::SnapshotComplete,
connector_state: None,
signal_acknowledgements: Vec::new(),
}))
.await
.map_err(|_| Error::Cancelled)?;
client.close().await.map_err(sqlserver_error)?;
Ok(anchor)
}
async fn current_anchor(&self) -> Result<Vec<u8>> {
let mut client = connect(&self.config, self.database()).await?;
let anchor = current_max_lsn(&mut client).await?;
client.close().await.map_err(sqlserver_error)?;
Ok(anchor)
}
fn signal_capture(&self) -> Option<CaptureTable> {
let signal = signal_table_key(&self.config)?;
self.captures
.iter()
.find(|capture| capture.key() == signal)
.cloned()
}
async fn emit_incremental_watermark(&self, id: &str, signal_type: &str) -> Result<()> {
let capture = self.signal_capture().ok_or_else(|| {
Error::Configuration(
"SQL Server incremental snapshots require a CDC-enabled signal.data.collection"
.into(),
)
})?;
let table = format!(
"{}.{}",
quote_identifier(&capture.schema),
quote_identifier(&capture.table)
);
let data = "{}";
let mut client = connect(&self.config, self.database()).await?;
client
.execute(
format!("INSERT INTO {table} ([id], [type], [data]) VALUES (@P1, @P2, @P3)"),
&[&id, &signal_type, &data],
)
.await
.map_err(sqlserver_error)?;
client.close().await.map_err(sqlserver_error)
}
async fn validate_incremental_schema(&self, capture: &CaptureTable) -> Result<()> {
let mut client = connect(&self.config, self.database()).await?;
let fields = discover_fields(&mut client, capture.source_object_id).await?;
client.close().await.map_err(sqlserver_error)?;
if fields != capture.event_schema.fields {
return Err(Error::Source(format!(
"SQL Server schema changed while incremental snapshot window for {}.{} was open",
capture.schema, capture.table
)));
}
Ok(())
}
fn capture_for_collection(&self, collection: &str) -> Option<CaptureTable> {
self.captures
.iter()
.find(|capture| format!("{}.{}", capture.schema, capture.table) == collection)
.cloned()
}
fn expand_data_collections(&self, patterns: &[String]) -> Result<Vec<String>> {
let patterns = compile_collection_patterns(patterns)?;
let signal_table = signal_table_key(&self.config);
let mut collections = self
.captures
.iter()
.filter(|capture| signal_table.as_ref() != Some(&capture.key()))
.filter_map(|capture| {
let short = format!("{}.{}", capture.schema, capture.table);
let qualified = format!("{}.{}", self.database(), short);
patterns
.iter()
.any(|pattern| pattern.is_match(&short) || pattern.is_match(&qualified))
.then_some(short)
})
.collect::<Vec<_>>();
collections.sort();
collections.dedup();
if collections.is_empty() {
return Err(Error::Source(
"SQL Server incremental snapshot patterns select no captured tables".into(),
));
}
Ok(collections)
}
async fn read_incremental_chunk(
&self,
capture: &CaptureTable,
progress: &IncrementalSnapshotProgress,
) -> Result<IncrementalChunk> {
let key_fields = capture
.event_schema
.fields
.iter()
.enumerate()
.filter(|(_, field)| field.primary_key)
.collect::<Vec<_>>();
if key_fields.is_empty() {
return Err(Error::Source(format!(
"SQL Server incremental snapshot table {}.{} requires a primary key",
capture.schema, capture.table
)));
}
let key_indices = key_fields
.iter()
.map(|(index, _)| *index)
.collect::<Vec<_>>();
let key_columns = key_fields
.iter()
.map(|(_, field)| format!("ct.{}", quote_identifier(&field.name)))
.collect::<Vec<_>>();
let collection = format!("{}.{}", capture.schema, capture.table);
let qualified_collection = format!("{}.{}", self.database(), collection);
let condition = progress
.additional_conditions
.iter()
.find_map(|(pattern, filter)| {
regex::Regex::new(&format!(r"^(?:{pattern})$"))
.ok()
.and_then(|pattern| {
(pattern.is_match(&collection) || pattern.is_match(&qualified_collection))
.then_some(filter)
})
});
let table = format!(
"{}.{}",
quote_identifier(&capture.schema),
quote_identifier(&capture.table)
);
let mut client = connect(&self.config, self.database()).await?;
let current_fields = discover_fields(&mut client, capture.source_object_id).await?;
if current_fields != capture.event_schema.fields {
return Err(Error::Source(format!(
"SQL Server schema changed before incremental snapshot query for {}.{}",
capture.schema, capture.table
)));
}
let maximum_key = match &progress.maximum_key {
Some(key) => Some(key.clone()),
None => {
let projections = key_fields
.iter()
.enumerate()
.map(|(projection_index, (_, field))| {
change_value_expression(projection_index, field)
})
.collect::<Vec<_>>()
.join(", ");
let where_clause = condition
.map(|condition| format!(" WHERE ({condition})"))
.unwrap_or_default();
let ordering = key_columns
.iter()
.map(|column| format!("{column} DESC"))
.collect::<Vec<_>>()
.join(", ");
let row = client
.simple_query(format!(
"SELECT TOP (1) {projections} FROM {table} AS ct{where_clause} ORDER BY {ordering}"
))
.await
.map_err(sqlserver_error)?
.into_row()
.await
.map_err(sqlserver_error)?;
row.map(|row| sqlserver_key_from_projection(&row, &key_fields))
.transpose()?
}
};
let Some(maximum_key) = maximum_key else {
client.close().await.map_err(sqlserver_error)?;
return Ok(IncrementalChunk {
rows: Vec::new(),
maximum_key: None,
});
};
validate_key_width(&key_columns, &maximum_key, "maximum")?;
let mut predicates = Vec::new();
if let Some(condition) = condition {
predicates.push(format!("({condition})"));
}
if let Some(last_key) = &progress.last_key {
validate_key_width(&key_columns, last_key, "last")?;
predicates.push(sqlserver_key_predicate(&key_columns, last_key, true)?);
}
predicates.push(sqlserver_key_predicate(&key_columns, &maximum_key, false)?);
let projections = capture
.event_schema
.fields
.iter()
.enumerate()
.map(|(index, field)| change_value_expression(index, field))
.collect::<Vec<_>>()
.join(", ");
let query = format!(
"SELECT TOP ({}) {projections} FROM {table} AS ct WHERE {} ORDER BY {}",
self.config.incremental_snapshot_chunk_size,
predicates.join(" AND "),
key_columns.join(", ")
);
let rows = client
.simple_query(query)
.await
.map_err(sqlserver_error)?
.into_first_result()
.await
.map_err(sqlserver_error)?;
let rows = rows
.iter()
.map(|row| {
let mut converted = convert_tds_row(row, &capture.event_schema)?;
let key = key_indices
.iter()
.map(|index| {
let field = &capture.event_schema.fields[*index];
converted
.get(&field.name)
.ok_or_else(|| {
Error::Source(format!(
"SQL Server incremental row is missing key {:?}",
field.name
))
})
.and_then(sqlserver_key_from_data_value)
})
.collect::<Result<Vec<_>>>()?;
transform_row(
self.column_transformer
.as_ref()
.expect("validated transformer"),
&mut converted,
self.database(),
capture,
);
Ok((converted, key))
})
.collect::<Result<Vec<_>>>()?;
client.close().await.map_err(sqlserver_error)?;
Ok(IncrementalChunk {
rows,
maximum_key: Some(maximum_key),
})
}
}
#[async_trait]
impl SourceConnector for SqlServerSource {
fn source_type(&self) -> &'static str {
"sqlserver"
}
async fn validate(&mut self) -> Result<()> {
self.validate_source().await
}
async fn run(&mut self, mut context: SourceContext) -> Result<()> {
let column_transformer = self.column_transformer.clone().map_or_else(
|| ColumnTransformer::new(&self.config.column_transformations),
Ok,
)?;
self.column_transformer = Some(column_transformer);
let checkpoint = context.initial_checkpoint.clone();
let snapshot_needed = match self.snapshot.mode {
SnapshotMode::Never => false,
SnapshotMode::Initial | SnapshotMode::WhenNeeded => checkpoint
.as_ref()
.is_none_or(|checkpoint| !checkpoint.snapshot_completed),
};
let mut checkpoint_position = checkpoint
.as_ref()
.map(|checkpoint| checkpoint.source_position.clone());
if checkpoint_position
.as_ref()
.is_some_and(|position| !matches!(position, SourcePosition::SqlServer(_)))
{
return Err(Error::State(
"SQL Server connector cannot resume from another source checkpoint".into(),
));
}
let checkpoint_has_connector_state = !snapshot_needed
&& checkpoint
.as_ref()
.and_then(|checkpoint| checkpoint.connector_state.as_ref())
.is_some();
let (mut incremental_progress, mut completed_signal_ids) = if !snapshot_needed {
checkpoint
.as_ref()
.and_then(|checkpoint| checkpoint.connector_state.as_ref())
.map(decode_connector_state)
.transpose()?
.unwrap_or_default()
} else {
(None, Vec::new())
};
let (mut cursor, mut resume_position) = if snapshot_needed {
(
CdcCursor::at_snapshot(self.run_snapshot(&context.output).await?),
None,
)
} else if let Some(SourcePosition::SqlServer(position)) = &checkpoint_position {
if position.database != self.database() {
return Err(Error::State(format!(
"SQL Server checkpoint belongs to database {:?}, not {:?}",
position.database,
self.database()
)));
}
CdcCursor::from_position(position, checkpoint_has_connector_state)?
} else {
(CdcCursor::at_snapshot(self.current_anchor().await?), None)
};
let mut client = connect(&self.config, self.database()).await?;
if self.snapshot.mode == SnapshotMode::WhenNeeded && !snapshot_needed {
if let Err(error) =
validate_retention(&mut client, &self.captures, &cursor.commit_lsn).await
{
if matches!(&error, Error::State(_)) {
warn!(
connector = %self.connector_name,
%error,
"SQL Server checkpoint is older than CDC retention; taking a recovery snapshot"
);
client.close().await.map_err(sqlserver_error)?;
let anchor = self.run_snapshot(&context.output).await?;
cursor = CdcCursor::at_snapshot(anchor);
resume_position = None;
checkpoint_position = None;
incremental_progress = None;
completed_signal_ids.clear();
client = connect(&self.config, self.database()).await?;
} else {
return Err(error);
}
}
} else {
validate_retention(&mut client, &self.captures, &cursor.commit_lsn).await?;
}
let mut state = StreamingState::new(cursor, resume_position);
let mut incremental =
SqlServerIncrementalSnapshot::new(incremental_progress, completed_signal_ids);
let mut last_safe_position = checkpoint_position
.as_ref()
.filter(|position| {
matches!(position, SourcePosition::SqlServer(position) if !position.snapshot)
})
.cloned()
.unwrap_or_else(|| state.safe_position(self.database()));
let mut interval = tokio::time::interval(self.config.poll_interval);
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut heartbeat = heartbeat_timer(self.config.heartbeat_interval);
let mut file_signal_poll = file_signal_timer(&self.config);
let mut incremental_tick = tokio::time::interval(std::time::Duration::from_millis(1));
incremental_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut heartbeat_connection = if self.config.heartbeat_interval.is_zero()
|| self.config.heartbeat_action_query.is_none()
{
None
} else {
Some(connect(&self.config, self.database()).await?)
};
info!(
connector = %self.connector_name,
database = %self.database(),
commit_lsn = %format_lsn(&state.cursor.commit_lsn),
max_retries = self.retry_policy.max_retries,
initial_retry_delay_ms = self.retry_policy.initial_delay.as_millis(),
max_retry_delay_ms = self.retry_policy.max_delay.as_millis(),
"SQL Server CDC streaming started"
);
loop {
tokio::select! {
_ = context.cancellation.cancelled() => {
client.close().await.map_err(sqlserver_error)?;
if let Some(connection) = heartbeat_connection {
connection.close().await.map_err(sqlserver_error)?;
}
return Ok(());
}
changed = context.acknowledged.changed() => {
if changed.is_err() {
return Err(Error::Cancelled);
}
}
_ = incremental_tick.tick(),
if incremental.is_active()
&& state.transaction.is_none()
&& state.cursor.commit_complete() => {
incremental.start_next_chunk(self).await?;
}
() = next_file_signal_poll(&mut file_signal_poll),
if signal_channel_enabled(&self.config, "file")
&& state.transaction.is_none()
&& state.cursor.commit_complete() => {
for line in crate::file_signal::read_and_clear(&self.config.signal_file).await? {
let record = match serde_json::from_str::<SignalRecord>(&line) {
Ok(record) => record,
Err(error) => {
tracing::warn!(%error, "invalid SQL Server file signal ignored");
continue;
}
};
let signal = match SqlServerIncrementalSnapshot::parse_external_record(&record) {
Ok(signal) => signal,
Err(error) => {
tracing::warn!(%error, "invalid SQL Server file signal ignored");
continue;
}
};
incremental.handle_signal(signal, self)?;
}
if incremental.state_dirty() {
emit_incremental_checkpoint(
&context.output,
incremental.progress(),
incremental.completed_signal_ids(),
&last_safe_position,
None,
).await?;
incremental.mark_checkpointed();
}
}
delivery = context.signals.recv(),
if (signal_channel_enabled(&self.config, "in-process")
|| signal_channel_enabled(&self.config, "kafka"))
&& state.transaction.is_none()
&& state.cursor.commit_complete() => {
let delivery = delivery.ok_or_else(|| {
Error::Source("SQL Server runtime signal channel closed".into())
})?;
let signal = match SqlServerIncrementalSnapshot::parse_external_record(delivery.record()) {
Ok(signal) => signal,
Err(error) => {
tracing::warn!(%error, "invalid SQL Server runtime signal ignored");
delivery.acknowledge();
continue;
}
};
incremental.handle_signal(signal, self)?;
emit_incremental_checkpoint(
&context.output,
incremental.progress(),
incremental.completed_signal_ids(),
&last_safe_position,
delivery.into_acknowledgement(),
).await?;
incremental.mark_checkpointed();
}
_ = interval.tick() => {
let Some(records) = poll_change_batch_with_retry(
self,
&mut client,
&mut state,
&context.cancellation,
).await? else {
continue;
};
for mut record in records {
if let Some(signal_row) = source_signal_row(&record, &self.config) {
if let Some(signal_row) = signal_row {
if incremental
.handle_internal_signal(&record, signal_row, self)
.await?
{
continue;
}
if signal_channel_enabled(&self.config, "source") {
let signal =
SqlServerIncrementalSnapshot::parse_row(signal_row)?;
incremental.handle_signal(signal, self)?;
}
}
continue;
}
incremental.observe_record(&record)?;
if incremental.closes_at(&record) {
let base = record.position.clone();
last_safe_position = incremental
.finish_window(self, &base, &context.output)
.await?;
continue;
}
if record.boundary == RecordBoundary::TransactionCommit
&& incremental.state_dirty()
{
record.connector_state = Some(encode_connector_state(
incremental.progress(),
incremental.completed_signal_ids(),
)?);
}
let checkpointed = record.connector_state.is_some();
let position = record.position.clone();
context.output.send(Ok(record)).await.map_err(|_| Error::Cancelled)?;
if position.is_after(&last_safe_position)
|| position == last_safe_position
{
last_safe_position = position;
}
if checkpointed {
incremental.mark_checkpointed();
}
}
}
() = next_heartbeat(&mut heartbeat) => {
if !state.cursor.commit_complete() {
continue;
}
if let Some(query) = self.config.heartbeat_action_query.clone() {
let connection = heartbeat_connection.take().ok_or_else(|| {
Error::Source("SQL Server heartbeat action connection is unavailable".into())
})?;
heartbeat_connection = Some(execute_heartbeat_action(connection, query).await?);
}
context.output.send(Ok(sqlserver_heartbeat_record(
&self.connector_name,
self.database(),
last_safe_position.clone(),
))).await.map_err(|_| Error::Cancelled)?;
}
}
}
}
}
async fn poll_change_batch_with_retry(
source: &SqlServerSource,
client: &mut SqlClient,
state: &mut StreamingState,
cancellation: &tokio_util::sync::CancellationToken,
) -> Result<Option<Vec<SourceRecord>>> {
let mut retries = 0_u64;
let mut delay = source.retry_policy.initial_delay;
let mut reconnect = false;
loop {
if reconnect {
match connect(&source.config, source.database()).await {
Ok(reconnected) => {
*client = reconnected;
info!(
connector = %source.connector_name,
database = %source.database(),
retries,
"SQL Server CDC polling connection recovered"
);
}
Err(error @ Error::RetryableSource(_)) => {
if !wait_for_sqlserver_retry(source, retries, &mut delay, &error, cancellation)
.await?
{
return Err(error);
}
retries += 1;
continue;
}
Err(error) => return Err(error),
}
}
match poll_change_batch_once(source, client, state).await {
Ok(records) => return Ok(records),
Err(error @ Error::RetryableSource(_)) => {
if !wait_for_sqlserver_retry(source, retries, &mut delay, &error, cancellation)
.await?
{
return Err(error);
}
retries += 1;
reconnect = true;
}
Err(error) => return Err(error),
}
}
}
async fn wait_for_sqlserver_retry(
source: &SqlServerSource,
retries: u64,
delay: &mut std::time::Duration,
error: &Error,
cancellation: &tokio_util::sync::CancellationToken,
) -> Result<bool> {
if !sqlserver_retry_allowed(&source.retry_policy, retries) {
return Ok(false);
}
warn!(
connector = %source.connector_name,
database = %source.database(),
retry = retries + 1,
max_retries = source.retry_policy.max_retries,
delay_ms = delay.as_millis(),
%error,
"retryable SQL Server CDC polling failure; reconnecting"
);
tokio::select! {
() = cancellation.cancelled() => return Err(Error::Cancelled),
() = tokio::time::sleep(*delay) => {}
}
*delay = delay.saturating_mul(2).min(source.retry_policy.max_delay);
Ok(true)
}
fn sqlserver_retry_allowed(policy: &RetryPolicy, retries: u64) -> bool {
policy.max_retries < 0 || retries < policy.max_retries as u64
}
async fn poll_change_batch_once(
source: &SqlServerSource,
client: &mut SqlClient,
state: &mut StreamingState,
) -> Result<Option<Vec<SourceRecord>>> {
let max_lsn = current_max_lsn(client).await?;
if max_lsn < state.cursor.commit_lsn
|| (max_lsn == state.cursor.commit_lsn && state.cursor.commit_complete())
{
return Ok(None);
}
validate_retention(client, &source.captures, &state.cursor.commit_lsn).await?;
read_change_batch(
client,
source.database(),
&source.connector_name,
&source.captures,
state,
&max_lsn,
source.config.streaming_fetch_size,
source
.column_transformer
.as_ref()
.expect("validated transformer"),
signal_table_key(&source.config).as_ref(),
)
.await
.map(Some)
}
fn heartbeat_timer(interval: std::time::Duration) -> Option<tokio::time::Interval> {
if interval.is_zero() {
return None;
}
let mut timer = tokio::time::interval_at(tokio::time::Instant::now() + interval, interval);
timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
Some(timer)
}
async fn next_heartbeat(timer: &mut Option<tokio::time::Interval>) {
match timer {
Some(timer) => {
timer.tick().await;
}
None => std::future::pending::<()>().await,
}
}
fn signal_channel_enabled(config: &SqlServerSourceConfig, channel: &str) -> bool {
config
.signal_enabled_channels
.iter()
.any(|enabled| enabled == channel)
}
fn signal_table_key(config: &SqlServerSourceConfig) -> Option<(String, String)> {
let parts = config
.signal_data_collection
.as_deref()?
.split('.')
.collect::<Vec<_>>();
match parts.as_slice() {
[schema, table] | [_, schema, table] => Some(((*schema).into(), (*table).into())),
_ => None,
}
}
fn transform_event(
transformer: &ColumnTransformer,
event: &mut ChangeEvent,
database: &str,
capture: &CaptureTable,
) {
transformer.transform_event_with_qualified_tables(
event,
&qualified_table_candidates(database, capture),
&HashMap::new(),
);
}
fn transform_row(
transformer: &ColumnTransformer,
row: &mut Row,
database: &str,
capture: &CaptureTable,
) {
transformer.transform_row_with_qualified_tables(
row,
&qualified_table_candidates(database, capture),
&capture.event_schema,
&HashMap::new(),
);
}
fn qualified_table_candidates(database: &str, capture: &CaptureTable) -> [String; 2] {
[
format!("{database}.{}.{}", capture.schema, capture.table),
format!("{}.{}", capture.schema, capture.table),
]
}
fn snapshot_includes(snapshot: &SnapshotConfig, database: &str, schema: &str, table: &str) -> bool {
snapshot.includes_collection(&format!("{database}.{schema}.{table}"))
}
fn validate_signal_schema(capture: &CaptureTable) -> Result<()> {
let expected = ["id", "type", "data"];
if capture.event_schema.fields.len() != expected.len()
|| capture
.event_schema
.fields
.iter()
.zip(expected)
.any(|(field, expected)| {
field.name != expected
|| !matches!(
base_type(&field.type_name),
"char" | "varchar" | "nchar" | "nvarchar" | "text" | "ntext"
)
})
{
return Err(Error::Configuration(format!(
"SQL Server signal table {}.{} must contain exactly text-compatible id, type, and data columns in that order",
capture.schema, capture.table
)));
}
let minimum_lengths = [
uuid::Uuid::nil().to_string().len() + "-close".len(),
WINDOW_OPEN_SIGNAL.len().max(WINDOW_CLOSE_SIGNAL.len()),
2,
];
for (field, minimum) in capture.event_schema.fields.iter().zip(minimum_lengths) {
if signal_text_capacity(&field.type_name).is_some_and(|capacity| capacity < minimum) {
return Err(Error::Configuration(format!(
"SQL Server signal table column {:?} must hold at least {minimum} characters for incremental snapshot watermarks; found {}",
field.name, field.type_name
)));
}
}
Ok(())
}
fn signal_text_capacity(type_name: &str) -> Option<usize> {
let normalized = type_name.trim().to_ascii_lowercase();
if matches!(base_type(&normalized), "text" | "ntext") || normalized.ends_with("(max)") {
return None;
}
normalized
.split_once('(')
.and_then(|(_, length)| length.strip_suffix(')'))
.and_then(|length| length.parse().ok())
}
async fn validate_signal_insert_permission(
client: &mut SqlClient,
capture: &CaptureTable,
) -> Result<()> {
let object_name = format!("{}.{}", capture.schema, capture.table);
let row = client
.query(
"SELECT CAST(HAS_PERMS_BY_NAME(@P1, 'OBJECT', 'INSERT') AS int) AS can_insert",
&[&object_name],
)
.await
.map_err(sqlserver_error)?
.into_row()
.await
.map_err(sqlserver_error)?
.ok_or_else(|| {
Error::Source("SQL Server did not return signal-table permissions".into())
})?;
if required::<i32>(&row, "can_insert")? != 1 {
return Err(Error::Configuration(format!(
"SQL Server connector user requires INSERT on signal table {}.{} for incremental snapshot watermarks",
capture.schema, capture.table
)));
}
Ok(())
}
fn source_signal_row<'a>(
record: &'a SourceRecord,
config: &SqlServerSourceConfig,
) -> Option<Option<&'a Row>> {
let signal = signal_table_key(config)?;
let event = record.event.as_ref()?;
if event.source.schema.as_deref() != Some(signal.0.as_str())
|| event.source.table.as_deref() != Some(signal.1.as_str())
{
return None;
}
Some(
(event.operation == Operation::Create)
.then_some(event.after.as_ref())
.flatten(),
)
}
fn file_signal_timer(config: &SqlServerSourceConfig) -> Option<tokio::time::Interval> {
if !signal_channel_enabled(config, "file") {
return None;
}
let mut timer = tokio::time::interval_at(
tokio::time::Instant::now() + config.signal_poll_interval,
config.signal_poll_interval,
);
timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
Some(timer)
}
async fn next_file_signal_poll(timer: &mut Option<tokio::time::Interval>) {
match timer {
Some(timer) => {
timer.tick().await;
}
None => std::future::pending::<()>().await,
}
}
fn signal_text(row: &Row, name: &str) -> Result<String> {
match row.get(name) {
Some(DataValue::String(value)) => Ok(value.clone()),
Some(DataValue::Json(value)) => Ok(value.to_string()),
Some(DataValue::Bytes(value)) => String::from_utf8(value.clone()).map_err(|error| {
Error::Source(format!(
"SQL Server signal column {name} is not UTF-8: {error}"
))
}),
Some(value) => Ok(value.to_json("__rustium_unavailable").to_string()),
None => Err(Error::Source(format!(
"SQL Server signal table is missing column {name:?}"
))),
}
}
fn parse_snapshot_signal(
id: &str,
signal_type: &str,
data: &serde_json::Value,
) -> Result<SnapshotSignal> {
match signal_type {
"execute-snapshot" => {
let snapshot_type = data.get("type").and_then(serde_json::Value::as_str);
if snapshot_type.is_some_and(|kind| !kind.eq_ignore_ascii_case("incremental")) {
return Err(Error::Source(format!(
"SQL Server execute-snapshot signal {id:?} supports only type=incremental"
)));
}
let collections = data_collections(data);
if collections.is_empty() {
return Err(Error::Source(format!(
"SQL Server execute-snapshot signal {id:?} has no data-collections"
)));
}
compile_collection_patterns(&collections)?;
let mut additional_conditions = BTreeMap::new();
if let Some(values) = data
.get("additional-conditions")
.and_then(serde_json::Value::as_array)
{
for value in values {
let collection = value
.get("data-collection")
.and_then(serde_json::Value::as_str)
.ok_or_else(|| {
Error::Source(format!(
"SQL Server execute-snapshot signal {id:?} has an invalid additional-condition"
))
})?;
let filter = value
.get("filter")
.and_then(serde_json::Value::as_str)
.ok_or_else(|| {
Error::Source(format!(
"SQL Server execute-snapshot signal {id:?} has an invalid additional-condition"
))
})?;
compile_collection_patterns(&[collection.into()])?;
additional_conditions.insert(collection.into(), filter.into());
}
}
Ok(SnapshotSignal::Execute {
id: id.into(),
data_collections: collections,
additional_conditions,
})
}
"stop-snapshot" => Ok(SnapshotSignal::Stop {
id: id.into(),
data_collections: data_collections(data),
}),
"pause-snapshot" => Ok(SnapshotSignal::Pause { id: id.into() }),
"resume-snapshot" => Ok(SnapshotSignal::Resume { id: id.into() }),
other => Ok(SnapshotSignal::Unsupported {
id: id.into(),
signal_type: other.into(),
}),
}
}
fn data_collections(data: &serde_json::Value) -> Vec<String> {
data.get("data-collections")
.and_then(serde_json::Value::as_array)
.map(|values| {
values
.iter()
.filter_map(serde_json::Value::as_str)
.map(str::to_owned)
.collect()
})
.unwrap_or_default()
}
fn compile_collection_patterns(patterns: &[String]) -> Result<Vec<regex::Regex>> {
patterns
.iter()
.map(|pattern| {
regex::Regex::new(&format!(r"^(?:{pattern})$")).map_err(|error| {
Error::Source(format!(
"invalid SQL Server snapshot collection pattern {pattern:?}: {error}"
))
})
})
.collect()
}
fn collection_matches_any(collection: &str, patterns: &[regex::Regex]) -> bool {
patterns.iter().any(|pattern| pattern.is_match(collection))
}
fn validate_key_width(columns: &[String], key: &[SqlServerKeyValue], boundary: &str) -> Result<()> {
if columns.len() != key.len() {
return Err(Error::State(format!(
"SQL Server incremental snapshot {boundary} key has {} values for {} primary-key columns",
key.len(),
columns.len()
)));
}
Ok(())
}
fn sqlserver_key_from_projection(
row: &TdsRow,
key_fields: &[(usize, &FieldSchema)],
) -> Result<Vec<SqlServerKeyValue>> {
key_fields
.iter()
.enumerate()
.map(|(projection_index, (_, field))| {
convert_tds_value(row, projection_index, &field.type_name)
.and_then(|value| sqlserver_key_from_data_value(&value))
})
.collect()
}
fn sqlserver_key_from_row(row: &Row, schema: &EventSchema) -> Result<Vec<SqlServerKeyValue>> {
let key = schema
.fields
.iter()
.filter(|field| field.primary_key)
.map(|field| {
row.get(&field.name)
.ok_or_else(|| {
Error::Source(format!(
"SQL Server CDC row is missing primary-key column {:?}",
field.name
))
})
.and_then(sqlserver_key_from_data_value)
})
.collect::<Result<Vec<_>>>()?;
if key.is_empty() {
return Err(Error::Source(
"SQL Server incremental snapshot table has no primary-key fields".into(),
));
}
Ok(key)
}
fn sqlserver_key_from_data_value(value: &DataValue) -> Result<SqlServerKeyValue> {
match value {
DataValue::Boolean(value) => Ok(SqlServerKeyValue::Boolean(*value)),
DataValue::Int32(value) => Ok(SqlServerKeyValue::Int32(*value)),
DataValue::Int64(value) => Ok(SqlServerKeyValue::Int64(*value)),
DataValue::UInt64(value) => Ok(SqlServerKeyValue::UInt64(*value)),
DataValue::Float64(value) if value.is_finite() => {
Ok(SqlServerKeyValue::Float64(value.to_bits()))
}
DataValue::Decimal(value) => Ok(SqlServerKeyValue::Decimal(value.clone())),
DataValue::String(value) => Ok(SqlServerKeyValue::String(value.clone())),
DataValue::Bytes(value) => Ok(SqlServerKeyValue::Bytes(value.clone())),
DataValue::Date(value) => Ok(SqlServerKeyValue::Date(value.clone())),
DataValue::Time(value) => Ok(SqlServerKeyValue::Time(value.clone())),
DataValue::Timestamp(value) => Ok(SqlServerKeyValue::Timestamp(value.clone())),
DataValue::Uuid(value) => Ok(SqlServerKeyValue::Uuid(*value)),
DataValue::Null => Err(Error::Source(
"SQL Server incremental snapshot primary key contains NULL".into(),
)),
other => Err(Error::Source(format!(
"SQL Server incremental snapshot primary key has unsupported value {other:?}"
))),
}
}
fn sqlserver_key_literal(value: &SqlServerKeyValue) -> Result<String> {
match value {
SqlServerKeyValue::Boolean(value) => Ok(u8::from(*value).to_string()),
SqlServerKeyValue::Int32(value) => Ok(value.to_string()),
SqlServerKeyValue::Int64(value) => Ok(value.to_string()),
SqlServerKeyValue::UInt64(value) => Ok(value.to_string()),
SqlServerKeyValue::Float64(bits) => {
let value = f64::from_bits(*bits);
if value.is_finite() {
Ok(value.to_string())
} else {
Err(Error::State(
"SQL Server incremental snapshot key contains a non-finite float".into(),
))
}
}
SqlServerKeyValue::Decimal(value) => {
let unsigned = value
.strip_prefix('-')
.or_else(|| value.strip_prefix('+'))
.unwrap_or(value);
let mut parts = unsigned.split('.');
let whole = parts.next().unwrap_or_default();
let fraction = parts.next();
if whole.is_empty()
|| !whole.chars().all(|character| character.is_ascii_digit())
|| fraction.is_some_and(|fraction| {
fraction.is_empty()
|| !fraction.chars().all(|character| character.is_ascii_digit())
})
|| parts.next().is_some()
{
return Err(Error::State(format!(
"SQL Server incremental snapshot has invalid decimal key {value:?}"
)));
}
Ok(value.clone())
}
SqlServerKeyValue::Bytes(value) => Ok(format!("0x{}", hex::encode(value))),
SqlServerKeyValue::String(value)
| SqlServerKeyValue::Date(value)
| SqlServerKeyValue::Time(value)
| SqlServerKeyValue::Timestamp(value) => Ok(format!("N'{}'", quote_literal(value))),
SqlServerKeyValue::Uuid(value) => Ok(format!("N'{value}'")),
}
}
fn sqlserver_key_predicate(
columns: &[String],
key: &[SqlServerKeyValue],
greater: bool,
) -> Result<String> {
validate_key_width(columns, key, if greater { "last" } else { "maximum" })?;
let literals = key
.iter()
.map(sqlserver_key_literal)
.collect::<Result<Vec<_>>>()?;
let mut terms = Vec::new();
for index in 0..columns.len() {
let mut comparisons = (0..index)
.map(|prefix| format!("{} = {}", columns[prefix], literals[prefix]))
.collect::<Vec<_>>();
comparisons.push(format!(
"{} {} {}",
columns[index],
if greater { ">" } else { "<" },
literals[index]
));
terms.push(format!("({})", comparisons.join(" AND ")));
}
if !greater {
terms.push(format!(
"({})",
columns
.iter()
.zip(&literals)
.map(|(column, literal)| format!("{column} = {literal}"))
.collect::<Vec<_>>()
.join(" AND ")
));
}
Ok(format!("({})", terms.join(" OR ")))
}
fn sqlserver_incremental_position(
base: &SourcePosition,
event_serial: u64,
) -> Result<SourcePosition> {
let SourcePosition::SqlServer(position) = base else {
return Err(Error::State(
"SQL Server incremental snapshot requires a SQL Server position".into(),
));
};
Ok(SourcePosition::SqlServer(SqlServerPosition {
database: position.database.clone(),
commit_lsn: position.commit_lsn.clone(),
change_lsn: position.change_lsn.clone(),
event_serial: position.event_serial.saturating_add(event_serial),
snapshot: false,
}))
}
async fn emit_incremental_checkpoint(
output: &tokio::sync::mpsc::Sender<Result<SourceRecord>>,
progress: Option<&IncrementalSnapshotProgress>,
completed_signal_ids: &[String],
position: &SourcePosition,
acknowledgement: Option<SignalAcknowledgement>,
) -> Result<()> {
output
.send(Ok(SourceRecord {
event: None,
position: position.clone(),
boundary: RecordBoundary::Heartbeat,
connector_state: Some(encode_connector_state(progress, completed_signal_ids)?),
signal_acknowledgements: acknowledgement.into_iter().collect(),
}))
.await
.map_err(|_| Error::Cancelled)
}
async fn execute_heartbeat_action(mut connection: SqlClient, query: String) -> Result<SqlClient> {
connection
.simple_query(&query)
.await
.map_err(sqlserver_error)?
.into_results()
.await
.map_err(sqlserver_error)?;
Ok(connection)
}
fn sqlserver_heartbeat_record(
connector_name: &str,
database: &str,
position: SourcePosition,
) -> SourceRecord {
let observed_time = Utc::now();
let mut attributes = BTreeMap::new();
attributes.insert("rustium.heartbeat".into(), true.into());
let mut after = Row::new();
after.insert(
"ts_ms".into(),
DataValue::Int64(observed_time.timestamp_millis()),
);
let event = ChangeEvent {
id: EventId::deterministic(
connector_name,
database,
&position,
"__heartbeat",
u64::try_from(observed_time.timestamp_micros()).unwrap_or_default(),
),
source: SourceMetadata {
connector: "sqlserver".into(),
connector_name: connector_name.into(),
database: database.into(),
schema: None,
table: None,
snapshot: false,
version: CONNECTOR_VERSION.into(),
attributes,
},
position: position.clone(),
transaction: None,
operation: Operation::Message,
before: None,
after: Some(after),
schema: EventSchema {
name: format!("{connector_name}.Heartbeat"),
version: 1,
fields: vec![FieldSchema {
name: "ts_ms".into(),
type_name: "int64".into(),
optional: false,
primary_key: false,
}],
},
source_time: None,
observed_time,
};
SourceRecord {
event: Some(event),
position,
boundary: RecordBoundary::Heartbeat,
connector_state: None,
signal_acknowledgements: Vec::new(),
}
}
struct StreamingState {
cursor: CdcCursor,
transaction: Option<ActiveTransaction>,
pending_updates: HashMap<(Vec<u8>, Vec<u8>, String), Row>,
resume_position: Option<SourcePosition>,
}
impl StreamingState {
fn new(cursor: CdcCursor, resume_position: Option<SourcePosition>) -> Self {
Self {
cursor,
transaction: None,
pending_updates: HashMap::new(),
resume_position,
}
}
fn transaction_metadata(
&mut self,
change: &RawChange,
table: &CaptureTable,
) -> TransactionMetadata {
if self
.transaction
.as_ref()
.is_none_or(|transaction| transaction.commit_lsn != change.commit_lsn)
{
self.transaction = Some(ActiveTransaction {
commit_lsn: change.commit_lsn.clone(),
source_time: change.source_time,
total_order: 0,
collection_order: HashMap::new(),
});
}
let transaction = self.transaction.as_mut().expect("transaction exists");
transaction.total_order += 1;
let collection_order = transaction.collection_order.entry(table.key()).or_insert(0);
*collection_order += 1;
TransactionMetadata {
id: format_lsn(&change.commit_lsn),
total_order: Some(transaction.total_order),
collection_order: Some(*collection_order),
}
}
fn commit_record(&mut self, database: &str, commit_lsn: &[u8]) -> Option<SourceRecord> {
self.transaction = None;
let record = SourceRecord {
event: None,
position: sqlserver_position(
database,
commit_lsn,
&max_lsn_bytes(),
COMMIT_SERIAL,
false,
),
boundary: RecordBoundary::TransactionCommit,
connector_state: None,
signal_acknowledgements: Vec::new(),
};
if self.should_skip(&record.position) {
None
} else {
Some(record)
}
}
fn should_skip(&mut self, position: &SourcePosition) -> bool {
let Some(resume) = &self.resume_position else {
return false;
};
if position.is_at_or_before(resume) {
true
} else {
self.resume_position = None;
false
}
}
fn safe_position(&self, database: &str) -> SourcePosition {
sqlserver_position(
database,
&self.cursor.commit_lsn,
&self.cursor.change_lsn,
COMMIT_SERIAL,
false,
)
}
}
#[allow(clippy::too_many_arguments)]
async fn read_change_batch(
client: &mut SqlClient,
database: &str,
connector_name: &str,
captures: &[CaptureTable],
state: &mut StreamingState,
max_lsn: &[u8],
fetch_size: usize,
column_transformer: &ColumnTransformer,
signal_table: Option<&(String, String)>,
) -> Result<Vec<SourceRecord>> {
let query = change_query(captures, fetch_size.saturating_add(1));
let rows = client
.query(
query,
&[
&state.cursor.commit_lsn,
&state.cursor.change_lsn,
&state.cursor.raw_operation,
&max_lsn.to_vec(),
],
)
.await
.map_err(sqlserver_error)?
.into_first_result()
.await
.map_err(sqlserver_error)?;
if rows.is_empty() {
if state.pending_updates.is_empty() {
state.cursor = CdcCursor::at_snapshot(max_lsn.to_vec());
return Ok(vec![SourceRecord {
event: None,
position: sqlserver_position(
database,
max_lsn,
&max_lsn_bytes(),
COMMIT_SERIAL,
false,
),
boundary: RecordBoundary::Heartbeat,
connector_state: None,
signal_acknowledgements: Vec::new(),
}]);
}
return Ok(Vec::new());
}
let has_lookahead = rows.len() > fetch_size;
let process_count = rows.len().min(fetch_size);
let mut raw_changes = Vec::with_capacity(process_count);
for row in rows.iter().take(process_count) {
raw_changes.push(raw_change(row, captures)?);
}
let lookahead_commit = has_lookahead
.then(|| required_binary(&rows[process_count], "commit_lsn"))
.transpose()?;
let complete_last_commit = lookahead_commit.as_ref().is_none_or(|lookahead| {
raw_changes
.last()
.is_some_and(|last| last.commit_lsn != *lookahead)
});
let capture_map = captures
.iter()
.map(|capture| (capture.capture_instance.as_str(), capture))
.collect::<HashMap<_, _>>();
let mut records = Vec::new();
for (index, change) in raw_changes.iter().enumerate() {
let table = capture_map
.get(change.capture_instance.as_str())
.copied()
.ok_or_else(|| {
Error::Source(format!(
"unknown SQL Server capture instance {:?}",
change.capture_instance
))
})?;
let update_key = (
change.commit_lsn.clone(),
change.change_lsn.clone(),
change.capture_instance.clone(),
);
let (operation, before, after, event_serial) = match change.raw_operation {
RAW_DELETE => (Operation::Delete, Some(change.row.clone()), None, 1),
RAW_INSERT => (Operation::Create, None, Some(change.row.clone()), 2),
RAW_UPDATE_BEFORE => {
state.pending_updates.insert(update_key, change.row.clone());
state.cursor = CdcCursor {
commit_lsn: change.commit_lsn.clone(),
change_lsn: change.change_lsn.clone(),
raw_operation: change.raw_operation,
};
continue;
}
RAW_UPDATE_AFTER => {
let before = state.pending_updates.remove(&update_key).ok_or_else(|| {
Error::Source(format!(
"SQL Server update after-image at {}:{} has no before-image",
format_lsn(&change.commit_lsn),
format_lsn(&change.change_lsn)
))
})?;
(Operation::Update, Some(before), Some(change.row.clone()), 4)
}
operation => {
return Err(Error::Source(format!(
"unsupported SQL Server CDC operation {operation}"
)));
}
};
let position = sqlserver_position(
database,
&change.commit_lsn,
&change.change_lsn,
event_serial,
false,
);
let transaction = state.transaction_metadata(change, table);
let source_time = state
.transaction
.as_ref()
.and_then(|transaction| transaction.source_time)
.or(change.source_time);
let mut event = ChangeEvent {
id: EventId::deterministic(
connector_name,
database,
&position,
&format!("{database}.{}.{}", table.schema, table.table),
event_serial,
),
source: source_metadata(connector_name, database, table, false),
position,
transaction: Some(transaction),
operation,
before,
after,
schema: table.event_schema.clone(),
source_time,
observed_time: Utc::now(),
};
if signal_table != Some(&table.key()) {
transform_event(column_transformer, &mut event, database, table);
}
let record = SourceRecord::data(event);
if !state.should_skip(&record.position) {
records.push(record);
}
state.cursor = CdcCursor {
commit_lsn: change.commit_lsn.clone(),
change_lsn: change.change_lsn.clone(),
raw_operation: change.raw_operation,
};
let next_commit = raw_changes.get(index + 1).map(|next| &next.commit_lsn);
let commit_complete = next_commit.is_some_and(|next| next != &change.commit_lsn)
|| (index + 1 == raw_changes.len() && complete_last_commit);
if commit_complete {
if let Some(record) = state.commit_record(database, &change.commit_lsn) {
records.push(record);
}
state.cursor = CdcCursor::at_snapshot(change.commit_lsn.clone());
}
}
Ok(records)
}
async fn connect(config: &SqlServerSourceConfig, database: &str) -> Result<SqlClient> {
let mut tds = TdsConfig::new();
tds.host(&config.hostname);
tds.port(config.port);
tds.database(database);
tds.application_name("rustium");
tds.authentication(AuthMethod::sql_server(&config.username, &config.password));
if config.encrypt {
tds.encryption(EncryptionLevel::Required);
} else {
tds.encryption(EncryptionLevel::NotSupported);
}
if config.trust_server_certificate {
tds.trust_cert();
}
let address = tds.get_addr();
let tcp = tokio::time::timeout(config.connect_timeout, TcpStream::connect(&address))
.await
.map_err(|_| Error::RetryableSource("timed out connecting to SQL Server".into()))?
.map_err(sqlserver_io_error)?;
tcp.set_nodelay(true).map_err(sqlserver_io_error)?;
tokio::time::timeout(
config.connect_timeout,
Client::connect(tds, tcp.compat_write()),
)
.await
.map_err(|_| Error::RetryableSource("timed out negotiating SQL Server TDS".into()))?
.map_err(sqlserver_error)
}
async fn discover_captures(
client: &mut SqlClient,
config: &SqlServerSourceConfig,
connector_name: &str,
) -> Result<Vec<CaptureTable>> {
let rows = client
.simple_query(
"SELECT s.name AS schema_name, t.name AS table_name, ct.capture_instance, \
CAST(ct.source_object_id AS int) AS source_object_id \
FROM cdc.change_tables ct \
JOIN sys.tables t ON t.object_id = ct.source_object_id \
JOIN sys.schemas s ON s.schema_id = t.schema_id \
ORDER BY s.name, t.name, ct.capture_instance",
)
.await
.map_err(sqlserver_error)?
.into_first_result()
.await
.map_err(sqlserver_error)?;
let mut captures = Vec::new();
let mut selected_tables = HashMap::new();
let signal_table = signal_table_key(config);
for row in rows {
let schema = required_string(&row, "schema_name")?;
let table = required_string(&row, "table_name")?;
let is_signal = signal_table.as_ref() == Some(&(schema.clone(), table.clone()));
if !config.tables.includes(&schema, &table) && !is_signal {
continue;
}
if selected_tables
.insert((schema.clone(), table.clone()), ())
.is_some()
{
return Err(Error::Configuration(format!(
"SQL Server table {schema}.{table} has multiple CDC capture instances; Rustium requires an explicit single active instance"
)));
}
let source_object_id = required::<i32>(&row, "source_object_id")?;
let fields = discover_fields(client, source_object_id).await?;
captures.push(CaptureTable {
schema: schema.clone(),
table: table.clone(),
capture_instance: required_string(&row, "capture_instance")?,
source_object_id,
event_schema: EventSchema {
name: format!("{connector_name}.{schema}.{table}.Envelope"),
version: 1,
fields,
},
});
}
Ok(captures)
}
async fn discover_fields(client: &mut SqlClient, object_id: i32) -> Result<Vec<FieldSchema>> {
let rows = client
.query(
"SELECT c.name AS column_name, ty.name AS type_name, CAST(c.max_length AS int) AS max_length, \
CAST(c.precision AS int) AS precision, CAST(c.scale AS int) AS scale, \
CAST(c.is_nullable AS bit) AS is_nullable, \
CAST(CASE WHEN pk.column_id IS NULL THEN 0 ELSE 1 END AS bit) AS is_primary_key \
FROM sys.columns c \
JOIN sys.types ty ON ty.user_type_id = c.user_type_id \
LEFT JOIN ( \
SELECT ic.object_id, ic.column_id FROM sys.indexes i \
JOIN sys.index_columns ic ON ic.object_id = i.object_id AND ic.index_id = i.index_id \
WHERE i.is_primary_key = 1 \
) pk ON pk.object_id = c.object_id AND pk.column_id = c.column_id \
WHERE c.object_id = @P1 ORDER BY c.column_id",
&[&object_id],
)
.await
.map_err(sqlserver_error)?
.into_first_result()
.await
.map_err(sqlserver_error)?;
let mut fields = Vec::new();
for row in rows {
let base = required_string(&row, "type_name")?;
fields.push(FieldSchema {
name: required_string(&row, "column_name")?,
type_name: format_sql_type(
&base,
required::<i32>(&row, "max_length")?,
required::<i32>(&row, "precision")?,
required::<i32>(&row, "scale")?,
),
optional: required::<bool>(&row, "is_nullable")?,
primary_key: required::<bool>(&row, "is_primary_key")?,
});
}
if fields.is_empty() {
return Err(Error::Source(format!(
"could not discover SQL Server columns for object_id {object_id}"
)));
}
Ok(fields)
}
async fn current_max_lsn(client: &mut SqlClient) -> Result<Vec<u8>> {
let row = client
.simple_query("SELECT sys.fn_cdc_get_max_lsn() AS max_lsn")
.await
.map_err(sqlserver_error)?
.into_row()
.await
.map_err(sqlserver_error)?
.ok_or_else(|| Error::Source("SQL Server returned no maximum CDC LSN".into()))?;
required_binary(&row, "max_lsn")
}
async fn validate_retention(
client: &mut SqlClient,
captures: &[CaptureTable],
restart_lsn: &[u8],
) -> Result<()> {
for capture in captures {
let row = client
.query(
"SELECT sys.fn_cdc_get_min_lsn(@P1) AS min_lsn",
&[&capture.capture_instance],
)
.await
.map_err(sqlserver_error)?
.into_row()
.await
.map_err(sqlserver_error)?
.ok_or_else(|| Error::Source("SQL Server returned no minimum CDC LSN".into()))?;
let min_lsn = required_binary(&row, "min_lsn")?;
if restart_lsn != zero_lsn_bytes() && restart_lsn < min_lsn.as_slice() {
return Err(Error::State(format!(
"SQL Server CDC cleanup advanced capture instance {:?} to {}; checkpoint {} is no longer available",
capture.capture_instance,
format_lsn(&min_lsn),
format_lsn(restart_lsn)
)));
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn snapshot_table(
client: &mut SqlClient,
database: &str,
connector_name: &str,
capture: &CaptureTable,
anchor: &[u8],
ordinal: &mut u64,
output: &tokio::sync::mpsc::Sender<Result<SourceRecord>>,
column_transformer: &ColumnTransformer,
) -> Result<()> {
let values = capture
.event_schema
.fields
.iter()
.enumerate()
.map(|(index, field)| change_value_expression(index, field))
.collect::<Vec<_>>()
.join(", ");
let primary_key = capture
.event_schema
.fields
.iter()
.filter(|field| field.primary_key)
.map(|field| format!("ct.{}", quote_identifier(&field.name)))
.collect::<Vec<_>>();
let ordering = if primary_key.is_empty() {
String::new()
} else {
format!(" ORDER BY {}", primary_key.join(", "))
};
let query = format!(
"SELECT {values} FROM {}.{} AS ct{ordering}",
quote_identifier(&capture.schema),
quote_identifier(&capture.table)
);
let mut rows = client
.simple_query(query)
.await
.map_err(sqlserver_error)?
.into_row_stream();
let mut count = 0_u64;
while let Some(row) = rows.try_next().await.map_err(sqlserver_error)? {
*ordinal += 1;
count += 1;
let position = sqlserver_position(database, anchor, &zero_lsn_bytes(), *ordinal, true);
let mut event = ChangeEvent {
id: EventId::deterministic(
connector_name,
database,
&position,
&format!("{database}.{}.{}", capture.schema, capture.table),
*ordinal,
),
source: source_metadata(connector_name, database, capture, true),
position,
transaction: None,
operation: Operation::Read,
before: None,
after: Some(convert_tds_row(&row, &capture.event_schema)?),
schema: capture.event_schema.clone(),
source_time: None,
observed_time: Utc::now(),
};
transform_event(column_transformer, &mut event, database, capture);
output
.send(Ok(SourceRecord::data(event)))
.await
.map_err(|_| Error::Cancelled)?;
}
drop(rows);
info!(table = %format!("{}.{}", capture.schema, capture.table), rows = count, "SQL Server snapshot table completed");
Ok(())
}
fn change_query(captures: &[CaptureTable], limit: usize) -> String {
let unions = captures
.iter()
.map(|capture| {
let values = capture
.event_schema
.fields
.iter()
.enumerate()
.map(|(index, field)| change_value_expression(index, field))
.collect::<Vec<_>>()
.join(", ");
format!(
"SELECT ct.__$start_lsn AS commit_lsn, ct.__$seqval AS change_lsn, \
CAST(ct.__$operation AS int) AS operation, \
CAST(N'{}' AS nvarchar(128)) AS capture_instance, \
(SELECT {values} FOR JSON PATH, WITHOUT_ARRAY_WRAPPER, INCLUDE_NULL_VALUES) AS row_data, \
sys.fn_cdc_map_lsn_to_time(ct.__$start_lsn) AS source_time \
FROM cdc.{} ct",
quote_literal(&capture.capture_instance),
quote_identifier(&format!("{}_CT", capture.capture_instance)),
)
})
.collect::<Vec<_>>()
.join(" UNION ALL ");
format!(
"SELECT TOP ({limit}) commit_lsn, change_lsn, operation, capture_instance, row_data, source_time \
FROM ({unions}) AS changes \
WHERE (commit_lsn > @P1 OR (commit_lsn = @P1 AND \
(change_lsn > @P2 OR (change_lsn = @P2 AND operation > @P3)))) \
AND commit_lsn <= @P4 \
ORDER BY commit_lsn, change_lsn, operation"
)
}
fn change_value_expression(index: usize, field: &FieldSchema) -> String {
let column = format!("ct.{}", quote_identifier(&field.name));
let base = base_type(&field.type_name);
let value = match base {
"binary" | "varbinary" | "image" | "rowversion" | "timestamp" => {
format!("CONVERT(varchar(max), {column}, 2)")
}
"date" | "time" | "datetime" | "datetime2" | "smalldatetime" | "datetimeoffset" => {
format!("CONVERT(nvarchar(64), {column}, 126)")
}
"geometry" | "geography" => {
format!("CONVERT(varchar(max), {column}.Serialize(), 2)")
}
"hierarchyid" => format!("{column}.ToString()"),
_ => format!("CONVERT(nvarchar(max), {column})"),
};
format!("{value} AS [c{index}]")
}
fn raw_change(row: &TdsRow, captures: &[CaptureTable]) -> Result<RawChange> {
let capture_instance = required_string(row, "capture_instance")?;
let capture = captures
.iter()
.find(|capture| capture.capture_instance == capture_instance)
.ok_or_else(|| {
Error::Source(format!(
"SQL Server returned unknown capture instance {capture_instance:?}"
))
})?;
let json = required_string(row, "row_data")?;
Ok(RawChange {
commit_lsn: required_binary(row, "commit_lsn")?,
change_lsn: required_binary(row, "change_lsn")?,
raw_operation: required::<i32>(row, "operation")?,
capture_instance,
row: convert_json_row(&json, &capture.event_schema)?,
source_time: row
.try_get::<NaiveDateTime, _>("source_time")
.map_err(sqlserver_error)?
.map(|time| DateTime::from_naive_utc_and_offset(time, Utc)),
})
}
fn convert_json_row(json: &str, schema: &EventSchema) -> Result<Row> {
let value: serde_json::Value = serde_json::from_str(json)?;
let object = value
.as_object()
.ok_or_else(|| Error::Source("SQL Server CDC row JSON is not an object".into()))?;
Ok(schema
.fields
.iter()
.enumerate()
.map(|(index, field)| {
let value = object
.get(&format!("c{index}"))
.map_or(DataValue::Null, |value| {
json_text_value(value, &field.type_name)
});
(field.name.clone(), value)
})
.collect())
}
fn json_text_value(value: &serde_json::Value, type_name: &str) -> DataValue {
match value {
serde_json::Value::Null => DataValue::Null,
serde_json::Value::String(value) => convert_sqlserver_text(value, type_name),
serde_json::Value::Bool(value) => DataValue::Boolean(*value),
serde_json::Value::Number(value) => DataValue::Decimal(value.to_string()),
_ => DataValue::Json(value.clone()),
}
}
fn convert_sqlserver_text(value: &str, type_name: &str) -> DataValue {
match base_type(type_name) {
"bit" => DataValue::Boolean(matches!(value, "1" | "true" | "TRUE")),
"tinyint" | "smallint" | "int" => value
.parse::<i32>()
.map_or_else(|_| DataValue::String(value.into()), DataValue::Int32),
"bigint" => value
.parse::<i64>()
.map_or_else(|_| DataValue::String(value.into()), DataValue::Int64),
"real" | "float" => value
.parse::<f64>()
.map_or_else(|_| DataValue::String(value.into()), DataValue::Float64),
"decimal" | "numeric" | "money" | "smallmoney" => DataValue::Decimal(value.into()),
"binary" | "varbinary" | "image" | "rowversion" | "timestamp" | "geometry"
| "geography" => {
hex::decode(value).map_or_else(|_| DataValue::String(value.into()), DataValue::Bytes)
}
"date" => DataValue::Date(value.into()),
"time" => DataValue::Time(value.into()),
"datetime" | "datetime2" | "smalldatetime" | "datetimeoffset" => {
DataValue::Timestamp(value.into())
}
"uniqueidentifier" => uuid::Uuid::parse_str(value)
.map_or_else(|_| DataValue::String(value.into()), DataValue::Uuid),
_ => DataValue::String(value.into()),
}
}
fn convert_tds_row(row: &TdsRow, schema: &EventSchema) -> Result<Row> {
if row.len() != schema.fields.len() {
return Err(Error::Source(format!(
"SQL Server snapshot row has {} values but schema has {} fields",
row.len(),
schema.fields.len()
)));
}
schema
.fields
.iter()
.enumerate()
.map(|(index, field)| {
Ok((
field.name.clone(),
convert_tds_value(row, index, &field.type_name)?,
))
})
.collect()
}
fn convert_tds_value(row: &TdsRow, index: usize, type_name: &str) -> Result<DataValue> {
let cell = row
.cells()
.nth(index)
.map(|(_, value)| value)
.ok_or_else(|| Error::Source("SQL Server row cell is missing".into()))?;
let value = match cell {
ColumnData::U8(value) => {
value.map_or(DataValue::Null, |value| DataValue::Int32(value.into()))
}
ColumnData::I16(value) => {
value.map_or(DataValue::Null, |value| DataValue::Int32(value.into()))
}
ColumnData::I32(value) => value.map_or(DataValue::Null, DataValue::Int32),
ColumnData::I64(value) => value.map_or(DataValue::Null, DataValue::Int64),
ColumnData::F32(value) => {
value.map_or(DataValue::Null, |value| DataValue::Float64(value.into()))
}
ColumnData::F64(value) => value.map_or(DataValue::Null, DataValue::Float64),
ColumnData::Bit(value) => value.map_or(DataValue::Null, DataValue::Boolean),
ColumnData::String(value) => value.as_ref().map_or(DataValue::Null, |value| {
convert_sqlserver_text(value, type_name)
}),
ColumnData::Guid(value) => value.map_or(DataValue::Null, DataValue::Uuid),
ColumnData::Binary(value) => value
.as_ref()
.map_or(DataValue::Null, |value| DataValue::Bytes(value.to_vec())),
ColumnData::Numeric(value) => value.map_or(DataValue::Null, |value| {
DataValue::Decimal(value.to_string())
}),
ColumnData::Xml(value) => value.as_ref().map_or(DataValue::Null, |value| {
DataValue::String(value.to_string())
}),
ColumnData::DateTime(None)
| ColumnData::SmallDateTime(None)
| ColumnData::Time(None)
| ColumnData::Date(None)
| ColumnData::DateTime2(None)
| ColumnData::DateTimeOffset(None) => DataValue::Null,
ColumnData::DateTime(Some(_))
| ColumnData::SmallDateTime(Some(_))
| ColumnData::DateTime2(Some(_)) => row
.try_get::<NaiveDateTime, _>(index)
.map_err(sqlserver_error)?
.map_or(DataValue::Null, |value| {
DataValue::Timestamp(sqlserver_datetime(value))
}),
ColumnData::Time(Some(_)) => row
.try_get::<NaiveTime, _>(index)
.map_err(sqlserver_error)?
.map_or(DataValue::Null, |value| DataValue::Time(value.to_string())),
ColumnData::Date(Some(_)) => row
.try_get::<NaiveDate, _>(index)
.map_err(sqlserver_error)?
.map_or(DataValue::Null, |value| DataValue::Date(value.to_string())),
ColumnData::DateTimeOffset(Some(_)) => row
.try_get::<DateTime<FixedOffset>, _>(index)
.map_err(sqlserver_error)?
.map_or(DataValue::Null, |value| {
DataValue::Timestamp(value.to_rfc3339())
}),
};
Ok(value)
}
fn sqlserver_datetime(value: NaiveDateTime) -> String {
value.to_string().replacen(' ', "T", 1)
}
fn source_metadata(
connector_name: &str,
database: &str,
capture: &CaptureTable,
snapshot: bool,
) -> SourceMetadata {
let mut attributes = BTreeMap::new();
attributes.insert(
"capture_instance".into(),
capture.capture_instance.clone().into(),
);
attributes.insert("source_object_id".into(), capture.source_object_id.into());
SourceMetadata {
connector: "sqlserver".into(),
connector_name: connector_name.into(),
database: database.into(),
schema: Some(capture.schema.clone()),
table: Some(capture.table.clone()),
snapshot,
version: CONNECTOR_VERSION.into(),
attributes,
}
}
fn sqlserver_position(
database: &str,
commit_lsn: &[u8],
change_lsn: &[u8],
event_serial: u64,
snapshot: bool,
) -> SourcePosition {
SourcePosition::SqlServer(SqlServerPosition {
database: database.into(),
commit_lsn: format_lsn(commit_lsn),
change_lsn: format_lsn(change_lsn),
event_serial,
snapshot,
})
}
fn snapshot_begin_sql(mode: &str) -> &'static str {
match mode {
"exclusive" => "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE; BEGIN TRANSACTION",
"snapshot" => "SET TRANSACTION ISOLATION LEVEL SNAPSHOT; BEGIN TRANSACTION",
"read_committed" => "SET TRANSACTION ISOLATION LEVEL READ COMMITTED; BEGIN TRANSACTION",
"read_uncommitted" => "SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED; BEGIN TRANSACTION",
_ => "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ; BEGIN TRANSACTION",
}
}
fn format_sql_type(base: &str, max_length: i32, precision: i32, scale: i32) -> String {
match base {
"decimal" | "numeric" => format!("{base}({precision},{scale})"),
"char" | "varchar" | "binary" | "varbinary" => {
if max_length < 0 {
format!("{base}(max)")
} else {
format!("{base}({max_length})")
}
}
"nchar" | "nvarchar" => {
if max_length < 0 {
format!("{base}(max)")
} else {
format!("{base}({})", max_length / 2)
}
}
"datetime2" | "datetimeoffset" | "time" => format!("{base}({scale})"),
_ => base.into(),
}
}
fn required<'a, T>(row: &'a TdsRow, name: &str) -> Result<T>
where
T: tiberius::FromSql<'a>,
{
row.try_get(name)
.map_err(sqlserver_error)?
.ok_or_else(|| Error::Source(format!("SQL Server result {name:?} is null")))
}
fn required_string(row: &TdsRow, name: &str) -> Result<String> {
required::<&str>(row, name).map(str::to_string)
}
fn required_binary(row: &TdsRow, name: &str) -> Result<Vec<u8>> {
let value = required::<&[u8]>(row, name)?.to_vec();
if value.len() != LSN_SIZE {
return Err(Error::Source(format!(
"SQL Server {name} has {} bytes, expected {LSN_SIZE}",
value.len()
)));
}
Ok(value)
}
fn quote_identifier(identifier: &str) -> String {
format!("[{}]", identifier.replace(']', "]]"))
}
fn quote_literal(value: &str) -> String {
value.replace('\'', "''")
}
fn base_type(type_name: &str) -> &str {
type_name.split('(').next().unwrap_or(type_name)
}
fn format_lsn(lsn: &[u8]) -> String {
format!("0x{}", hex::encode_upper(lsn))
}
fn parse_lsn(lsn: &str) -> Result<Vec<u8>> {
let bytes = hex::decode(lsn.strip_prefix("0x").unwrap_or(lsn)).map_err(|error| {
Error::State(format!(
"invalid SQL Server checkpoint LSN {lsn:?}: {error}"
))
})?;
if bytes.len() != LSN_SIZE {
return Err(Error::State(format!(
"invalid SQL Server checkpoint LSN length {}; expected {LSN_SIZE}",
bytes.len()
)));
}
Ok(bytes)
}
fn zero_lsn_bytes() -> Vec<u8> {
vec![0; LSN_SIZE]
}
fn max_lsn_bytes() -> Vec<u8> {
vec![u8::MAX; LSN_SIZE]
}
fn sqlserver_error(error: tiberius::error::Error) -> Error {
let retryable = matches!(error, tiberius::error::Error::Io { .. })
|| error.code().is_some_and(is_transient_sqlserver_code);
if retryable {
Error::RetryableSource(error.to_string())
} else {
Error::Source(error.to_string())
}
}
fn sqlserver_io_error(error: std::io::Error) -> Error {
Error::RetryableSource(error.to_string())
}
const fn is_transient_sqlserver_code(code: u32) -> bool {
matches!(
code,
233 | 1205
| 1222
| 10_053
| 10_054
| 10_060
| 10_928
| 10_929
| 40_197
| 40_501
| 40_613
| 49_918
| 49_919
| 49_920
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn qualifies_sqlserver_snapshot_collection_filters() {
let snapshot = SnapshotConfig {
include_collections: vec![r"inventory\.dbo\.orders".into()],
..SnapshotConfig::default()
};
assert!(snapshot_includes(&snapshot, "inventory", "dbo", "orders"));
assert!(!snapshot_includes(&snapshot, "archive", "dbo", "orders"));
assert!(!snapshot_includes(
&snapshot,
"inventory",
"dbo",
"orders_history"
));
}
#[test]
fn round_trips_sqlserver_lsn() {
let bytes = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 255];
assert_eq!(parse_lsn(&format_lsn(&bytes)).unwrap(), bytes);
}
#[test]
fn distinguishes_complete_and_mid_transaction_cursors() {
let complete = CdcCursor::at_snapshot(vec![1; LSN_SIZE]);
assert!(complete.commit_complete());
let incomplete = CdcCursor {
commit_lsn: vec![1; LSN_SIZE],
change_lsn: vec![2; LSN_SIZE],
raw_operation: RAW_UPDATE_BEFORE,
};
assert!(!incomplete.commit_complete());
}
#[test]
fn classifies_only_transient_sqlserver_failures_for_retry() {
let io_error = tiberius::error::Error::Io {
kind: std::io::ErrorKind::ConnectionReset,
message: "connection reset".into(),
};
assert!(matches!(
sqlserver_error(io_error),
Error::RetryableSource(message) if message.contains("connection reset")
));
assert!(is_transient_sqlserver_code(1205));
assert!(is_transient_sqlserver_code(40_613));
assert!(!is_transient_sqlserver_code(208));
assert!(matches!(
sqlserver_error(tiberius::error::Error::Protocol("invalid token".into())),
Error::Source(message) if message.contains("invalid token")
));
}
#[test]
fn enforces_sqlserver_retry_policy_boundaries() {
let disabled = RetryPolicy {
max_retries: 0,
..RetryPolicy::default()
};
assert!(!sqlserver_retry_allowed(&disabled, 0));
let finite = RetryPolicy {
max_retries: 2,
..RetryPolicy::default()
};
assert!(sqlserver_retry_allowed(&finite, 0));
assert!(sqlserver_retry_allowed(&finite, 1));
assert!(!sqlserver_retry_allowed(&finite, 2));
let unbounded = RetryPolicy {
max_retries: -1,
..RetryPolicy::default()
};
assert!(sqlserver_retry_allowed(&unbounded, u64::MAX));
}
#[test]
fn converts_sqlserver_values() {
assert_eq!(
convert_sqlserver_text("12.30", "decimal(10,2)"),
DataValue::Decimal("12.30".into())
);
assert_eq!(convert_sqlserver_text("1", "bit"), DataValue::Boolean(true));
assert_eq!(
convert_sqlserver_text("00FF", "varbinary(2)"),
DataValue::Bytes(vec![0, 255])
);
assert_eq!(
convert_sqlserver_text("E6100000010C0000000000000040000000000000F03F", "geometry"),
DataValue::Bytes(hex::decode("E6100000010C0000000000000040000000000000F03F").unwrap())
);
assert_eq!(
change_value_expression(
0,
&FieldSchema {
name: "location".into(),
type_name: "geography".into(),
optional: false,
primary_key: false,
}
),
"CONVERT(varchar(max), ct.[location].Serialize(), 2) AS [c0]"
);
assert_eq!(
sqlserver_datetime(
NaiveDate::from_ymd_opt(2026, 7, 16)
.unwrap()
.and_hms_micro_opt(9, 30, 45, 123_400)
.unwrap()
),
"2026-07-16T09:30:45.123400"
);
}
#[test]
fn builds_bounded_direct_cdc_query() {
let capture = CaptureTable {
schema: "dbo".into(),
table: "orders".into(),
capture_instance: "dbo_orders".into(),
source_object_id: 42,
event_schema: EventSchema {
name: "orders".into(),
version: 1,
fields: vec![FieldSchema {
name: "id".into(),
type_name: "bigint".into(),
optional: false,
primary_key: true,
}],
},
};
let query = change_query(&[capture], 513);
assert!(query.contains("TOP (513)"));
assert!(query.contains("cdc.[dbo_orders_CT]"));
assert!(query.contains("ORDER BY commit_lsn, change_lsn, operation"));
}
#[test]
fn builds_composite_sqlserver_keyset_predicates() {
let columns = vec!["ct.[tenant_id]".into(), "ct.[id]".into()];
let key = vec![SqlServerKeyValue::Int32(7), SqlServerKeyValue::Int64(42)];
assert_eq!(
sqlserver_key_predicate(&columns, &key, true).unwrap(),
"((ct.[tenant_id] > 7) OR (ct.[tenant_id] = 7 AND ct.[id] > 42))"
);
assert_eq!(
sqlserver_key_predicate(&columns, &key, false).unwrap(),
"((ct.[tenant_id] < 7) OR (ct.[tenant_id] = 7 AND ct.[id] < 42) OR (ct.[tenant_id] = 7 AND ct.[id] = 42))"
);
}
#[test]
fn validates_sqlserver_signal_table_layout() {
let mut capture = CaptureTable {
schema: "dbo".into(),
table: "rustium_signal".into(),
capture_instance: "dbo_rustium_signal".into(),
source_object_id: 42,
event_schema: EventSchema {
name: "signal".into(),
version: 1,
fields: ["id", "type", "data"]
.into_iter()
.map(|name| FieldSchema {
name: name.into(),
type_name: "nvarchar(max)".into(),
optional: false,
primary_key: name == "id",
})
.collect(),
},
};
assert!(validate_signal_schema(&capture).is_ok());
capture.event_schema.fields[0].type_name = "nvarchar(20)".into();
assert!(matches!(
validate_signal_schema(&capture),
Err(Error::Configuration(message)) if message.contains("at least 42 characters")
));
}
#[test]
fn builds_sqlserver_heartbeat_record_at_commit_position() {
let position = sqlserver_position(
"inventory",
&[1; LSN_SIZE],
&max_lsn_bytes(),
COMMIT_SERIAL,
false,
);
let record =
sqlserver_heartbeat_record("inventory-sqlserver", "inventory", position.clone());
let event = record.event.unwrap();
assert_eq!(record.boundary, RecordBoundary::Heartbeat);
assert_eq!(record.position, position);
assert_eq!(event.operation, Operation::Message);
assert_eq!(event.source.table, None);
assert_eq!(
event.source.attributes.get("rustium.heartbeat"),
Some(&true.into())
);
assert!(matches!(
event.after.unwrap().get("ts_ms"),
Some(DataValue::Int64(_))
));
}
}