fix: normalize incremental cursor values in drivers before persisting state

drivers/mssql/internal/backfill.go

@@ -300,6 +300,11 @@ func normalizeBoundaryValue(value any, pkCols []string, columnType string) strin
 		return utils.ConvertToString(value)
 	}
 
+	value = normalizeMSSQLValue(value, columnType)
+	return utils.ConvertToString(value)
+}
+
+func normalizeMSSQLValue(value any, columnType string) any {
 	columnType = strings.ToLower(columnType)
 
 	switch v := value.(type) {
@@ -328,7 +333,8 @@ func normalizeBoundaryValue(value any, pkCols []string, columnType string) strin
 			return utils.HexEncode(v)
 		}
 	}
-	return utils.ConvertToString(value)
+
+	return value
 }
 
 // getTableExtremes returns MIN and MAX key values for the given PK columns

drivers/mssql/internal/incremental.go

@@ -49,5 +49,31 @@ func (m *MSSQL) FetchMaxCursorValues(ctx context.Context, stream types.StreamInt
 	if err != nil {
 		return nil, nil, err
 	}
+
+	primaryCursor, secondaryCursor := stream.Cursor()
+	maxPrimaryCursorValue, err = m.normalizeCursorValue(ctx, stream, primaryCursor, maxPrimaryCursorValue)
+	if err != nil {
+		return nil, nil, err
+	}
+	if secondaryCursor != "" {
+		maxSecondaryCursorValue, err = m.normalizeCursorValue(ctx, stream, secondaryCursor, maxSecondaryCursorValue)
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+
 	return maxPrimaryCursorValue, maxSecondaryCursorValue, nil
-}
+}
+
+func (m *MSSQL) normalizeCursorValue(ctx context.Context, stream types.StreamInterface, cursorField string, value any) (any, error) {
+	if value == nil {
+		return value, nil
+	}
+
+	var dataType string
+	if err := m.client.QueryRowContext(ctx, jdbc.MSSQLColumnTypeQuery(), stream.Namespace(), stream.Name(), cursorField).Scan(&dataType); err != nil {
+		return nil, fmt.Errorf("failed to get MSSQL cursor type: %s", err)
+	}
+
+	return normalizeMSSQLValue(value, dataType), nil
+}

pkg/jdbc/jdbc.go

@@ -1347,6 +1347,11 @@ func GetMaxCursorValues(ctx context.Context, client *sqlx.DB, driverType constan
 	var maxPrimaryCursorValue, maxSecondaryCursorValue any
 
 	bytesConverter := func(value any) any {
+		// MSSQL has type-aware normalization in the driver.
+		if driverType == constants.MSSQL {
+			return value
+		}
+
 		switch v := value.(type) {
 		case []byte:
 			return string(v)