lib/DBIx/Class/Storage/DBI/MSSQL.pm

   1 package DBIx::Class::Storage::DBI::MSSQL;
   2
   3 use strict;
   4 use warnings;
   5
   6 use base qw/DBIx::Class::Storage::DBI::UniqueIdentifier/;
   7 use mro 'c3';
   8 use Try::Tiny;
   9 use List::Util 'first';
  10 use namespace::clean;
  11
  12 __PACKAGE__->mk_group_accessors(simple => qw/
  13   _identity _identity_method _pre_insert_sql _post_insert_sql
  14 /);
  15
  16 __PACKAGE__->sql_maker_class('DBIx::Class::SQLMaker::MSSQL');
  17
  18 __PACKAGE__->sql_quote_char([qw/[ ]/]);
  19
  20 __PACKAGE__->datetime_parser_type (
  21   'DBIx::Class::Storage::DBI::MSSQL::DateTime::Format'
  22 );
  23
  24
  25 __PACKAGE__->new_guid('NEWID()');
  26
  27 sub _set_identity_insert {
  28   my ($self, $table) = @_;
  29
  30   my $stmt = 'SET IDENTITY_INSERT %s %s';
  31   $table   = $self->sql_maker->_quote($table);
  32
  33   $self->_pre_insert_sql (sprintf $stmt, $table, 'ON');
  34   $self->_post_insert_sql(sprintf $stmt, $table, 'OFF');
  35 }
  36
  37 sub insert_bulk {
  38   my $self = shift;
  39   my ($source, $cols, $data) = @_;
  40
  41   my $is_identity_insert =
  42     (first { $_->{is_auto_increment} } values %{ $source->columns_info($cols) } )
  43       ? 1
  44       : 0
  45   ;
  46
  47   if ($is_identity_insert) {
  48      $self->_set_identity_insert ($source->name);
  49   }
  50
  51   $self->next::method(@_);
  52 }
  53
  54 sub insert {
  55   my $self = shift;
  56   my ($source, $to_insert) = @_;
  57
  58   my $supplied_col_info = $self->_resolve_column_info($source, [keys %$to_insert] );
  59
  60   my $is_identity_insert =
  61     (first { $_->{is_auto_increment} } values %$supplied_col_info) ? 1 : 0;
  62
  63   if ($is_identity_insert) {
  64      $self->_set_identity_insert ($source->name);
  65   }
  66
  67   my $updated_cols = $self->next::method(@_);
  68
  69   return $updated_cols;
  70 }
  71
  72 sub _prep_for_execute {
  73   my $self = shift;
  74   my ($op, $extra_bind, $ident, $args) = @_;
  75
  76 # cast MONEY values properly
  77   if ($op eq 'insert' || $op eq 'update') {
  78     my $fields = $args->[0];
  79
  80     my $colinfo = $ident->columns_info([keys %$fields]);
  81
  82     for my $col (keys %$fields) {
  83       # $ident is a result source object with INSERT/UPDATE ops
  84       if (
  85         $colinfo->{$col}{data_type}
  86           &&
  87         $colinfo->{$col}{data_type} =~ /^money\z/i
  88       ) {
  89         my $val = $fields->{$col};
  90         $fields->{$col} = \['CAST(? AS MONEY)', [ $col => $val ]];
  91       }
  92     }
  93   }
  94
  95   my ($sql, $bind) = $self->next::method (@_);
  96
  97   if ($op eq 'insert') {
  98     if (my $prepend = $self->_pre_insert_sql) {
  99       $sql = "${prepend}\n${sql}";
 100       $self->_pre_insert_sql(undef);
 101     }
 102     if (my $append  = $self->_post_insert_sql) {
 103       $sql = "${sql}\n${append}";
 104       $self->_post_insert_sql(undef);
 105     }
 106     $sql .= "\nSELECT SCOPE_IDENTITY()";
 107   }
 108
 109   return ($sql, $bind);
 110 }
 111
 112 sub _execute {
 113   my $self = shift;
 114   my ($op) = @_;
 115
 116   my ($rv, $sth, @bind) = $self->dbh_do($self->can('_dbh_execute'), @_);
 117
 118   if ($op eq 'insert') {
 119
 120     # this should bring back the result of SELECT SCOPE_IDENTITY() we tacked
 121     # on in _prep_for_execute above
 122     my ($identity) = try { $sth->fetchrow_array };
 123
 124     # SCOPE_IDENTITY failed, but we can do something else
 125     if ( (! $identity) && $self->_identity_method) {
 126       ($identity) = $self->_dbh->selectrow_array(
 127         'select ' . $self->_identity_method
 128       );
 129     }
 130
 131     $self->_identity($identity);
 132     $sth->finish;
 133   }
 134
 135   return wantarray ? ($rv, $sth, @bind) : $rv;
 136 }
 137
 138 sub last_insert_id { shift->_identity }
 139
 140 #
 141 # MSSQL is retarded wrt ordered subselects. One needs to add a TOP
 142 # to *all* subqueries, but one also *can't* use TOP 100 PERCENT
 143 # http://sqladvice.com/forums/permalink/18496/22931/ShowThread.aspx#22931
 144 #
 145 sub _select_args_to_query {
 146   my $self = shift;
 147
 148   my ($sql, $prep_bind, @rest) = $self->next::method (@_);
 149
 150   # see if this is an ordered subquery
 151   my $attrs = $_[3];
 152   if (
 153     $sql !~ /^ \s* SELECT \s+ TOP \s+ \d+ \s+ /xi
 154       &&
 155     scalar $self->_extract_order_criteria ($attrs->{order_by})
 156   ) {
 157     $self->throw_exception(
 158       'An ordered subselect encountered - this is not safe! Please see "Ordered Subselects" in DBIx::Class::Storage::DBI::MSSQL
 159     ') unless $attrs->{unsafe_subselect_ok};
 160     my $max = $self->sql_maker->__max_int;
 161     $sql =~ s/^ \s* SELECT \s/SELECT TOP $max /xi;
 162   }
 163
 164   return wantarray
 165     ? ($sql, $prep_bind, @rest)
 166     : \[ "($sql)", @$prep_bind ]
 167   ;
 168 }
 169
 170
 171 # savepoint syntax is the same as in Sybase ASE
 172
 173 sub _svp_begin {
 174   my ($self, $name) = @_;
 175
 176   $self->_get_dbh->do("SAVE TRANSACTION $name");
 177 }
 178
 179 # A new SAVE TRANSACTION with the same name releases the previous one.
 180 sub _svp_release { 1 }
 181
 182 sub _svp_rollback {
 183   my ($self, $name) = @_;
 184
 185   $self->_get_dbh->do("ROLLBACK TRANSACTION $name");
 186 }
 187
 188 sub sqlt_type { 'SQLServer' }
 189
 190 sub sql_limit_dialect {
 191   my $self = shift;
 192
 193   my $supports_rno = 0;
 194
 195   if (exists $self->_server_info->{normalized_dbms_version}) {
 196     $supports_rno = 1 if $self->_server_info->{normalized_dbms_version} >= 9;
 197   }
 198   else {
 199     # User is connecting via DBD::Sybase and has no permission to run
 200     # stored procedures like xp_msver, or version detection failed for some
 201     # other reason.
 202     # So, we use a query to check if RNO is implemented.
 203     try {
 204       $self->_get_dbh->selectrow_array('SELECT row_number() OVER (ORDER BY rand())');
 205       $supports_rno = 1;
 206     };
 207   }
 208
 209   return $supports_rno ? 'RowNumberOver' : 'Top';
 210 }
 211
 212 sub _ping {
 213   my $self = shift;
 214
 215   my $dbh = $self->_dbh or return 0;
 216
 217   local $dbh->{RaiseError} = 1;
 218   local $dbh->{PrintError} = 0;
 219
 220   return try {
 221     $dbh->do('select 1');
 222     1;
 223   } catch {
 224     0;
 225   };
 226 }
 227
 228 package # hide from PAUSE
 229   DBIx::Class::Storage::DBI::MSSQL::DateTime::Format;
 230
 231 my $datetime_format      = '%Y-%m-%d %H:%M:%S.%3N'; # %F %T
 232 my $smalldatetime_format = '%Y-%m-%d %H:%M:%S';
 233
 234 my ($datetime_parser, $smalldatetime_parser);
 235
 236 sub parse_datetime {
 237   shift;
 238   require DateTime::Format::Strptime;
 239   $datetime_parser ||= DateTime::Format::Strptime->new(
 240     pattern  => $datetime_format,
 241     on_error => 'croak',
 242   );
 243   return $datetime_parser->parse_datetime(shift);
 244 }
 245
 246 sub format_datetime {
 247   shift;
 248   require DateTime::Format::Strptime;
 249   $datetime_parser ||= DateTime::Format::Strptime->new(
 250     pattern  => $datetime_format,
 251     on_error => 'croak',
 252   );
 253   return $datetime_parser->format_datetime(shift);
 254 }
 255
 256 sub parse_smalldatetime {
 257   shift;
 258   require DateTime::Format::Strptime;
 259   $smalldatetime_parser ||= DateTime::Format::Strptime->new(
 260     pattern  => $smalldatetime_format,
 261     on_error => 'croak',
 262   );
 263   return $smalldatetime_parser->parse_datetime(shift);
 264 }
 265
 266 sub format_smalldatetime {
 267   shift;
 268   require DateTime::Format::Strptime;
 269   $smalldatetime_parser ||= DateTime::Format::Strptime->new(
 270     pattern  => $smalldatetime_format,
 271     on_error => 'croak',
 272   );
 273   return $smalldatetime_parser->format_datetime(shift);
 274 }
 275
 276 1;
 277
 278 =head1 NAME
 279
 280 DBIx::Class::Storage::DBI::MSSQL - Base Class for Microsoft SQL Server support
 281 in DBIx::Class
 282
 283 =head1 SYNOPSIS
 284
 285 This is the base class for Microsoft SQL Server support, used by
 286 L<DBIx::Class::Storage::DBI::ODBC::Microsoft_SQL_Server> and
 287 L<DBIx::Class::Storage::DBI::Sybase::Microsoft_SQL_Server>.
 288
 289 =head1 IMPLEMENTATION NOTES
 290
 291 =head2 IDENTITY information
 292
 293 Microsoft SQL Server supports three methods of retrieving the IDENTITY
 294 value for inserted row: IDENT_CURRENT, @@IDENTITY, and SCOPE_IDENTITY().
 295 SCOPE_IDENTITY is used here because it is the safest.  However, it must
 296 be called is the same execute statement, not just the same connection.
 297
 298 So, this implementation appends a SELECT SCOPE_IDENTITY() statement
 299 onto each INSERT to accommodate that requirement.
 300
 301 C<SELECT @@IDENTITY> can also be used by issuing:
 302
 303   $self->_identity_method('@@identity');
 304
 305 it will only be used if SCOPE_IDENTITY() fails.
 306
 307 This is more dangerous, as inserting into a table with an on insert trigger that
 308 inserts into another table with an identity will give erroneous results on
 309 recent versions of SQL Server.
 310
 311 =head2 identity insert
 312
 313 Be aware that we have tried to make things as simple as possible for our users.
 314 For MSSQL that means that when a user tries to create a row, while supplying an
 315 explicit value for an autoincrementing column, we will try to issue the
 316 appropriate database call to make this possible, namely C<SET IDENTITY_INSERT
 317 $table_name ON>. Unfortunately this operation in MSSQL requires the
 318 C<db_ddladmin> privilege, which is normally not included in the standard
 319 write-permissions.
 320
 321 =head2 Ordered Subselects
 322
 323 If you attempted the following query (among many others) in Microsoft SQL
 324 Server
 325
 326  $rs->search ({}, {
 327   prefetch => 'relation',
 328   rows => 2,
 329   offset => 3,
 330  });
 331
 332 You may be surprised to receive an exception. The reason for this is a quirk
 333 in the MSSQL engine itself, and sadly doesn't have a sensible workaround due
 334 to the way DBIC is built. DBIC can do truly wonderful things with the aid of
 335 subselects, and does so automatically when necessary. The list of situations
 336 when a subselect is necessary is long and still changes often, so it can not
 337 be exhaustively enumerated here. The general rule of thumb is a joined
 338 L<has_many|DBIx::Class::Relationship/has_many> relationship with limit/group
 339 applied to the left part of the join.
 340
 341 In its "pursuit of standards" Microsft SQL Server goes to great lengths to
 342 forbid the use of ordered subselects. This breaks a very useful group of
 343 searches like "Give me things number 4 to 6 (ordered by name), and prefetch
 344 all their relations, no matter how many". While there is a hack which fools
 345 the syntax checker, the optimizer may B<still elect to break the subselect>.
 346 Testing has determined that while such breakage does occur (the test suite
 347 contains an explicit test which demonstrates the problem), it is relative
 348 rare. The benefits of ordered subselects are on the other hand too great to be
 349 outright disabled for MSSQL.
 350
 351 Thus compromise between usability and perfection is the MSSQL-specific
 352 L<resultset attribute|DBIx::Class::ResultSet/ATTRIBUTES> C<unsafe_subselect_ok>.
 353 It is deliberately not possible to set this on the Storage level, as the user
 354 should inspect (and preferably regression-test) the return of every such
 355 ResultSet individually. The example above would work if written like:
 356
 357  $rs->search ({}, {
 358   unsafe_subselect_ok => 1,
 359   prefetch => 'relation',
 360   rows => 2,
 361   offset => 3,
 362  });
 363
 364 If it is possible to rewrite the search() in a way that will avoid the need
 365 for this flag - you are urged to do so. If DBIC internals insist that an
 366 ordered subselect is necessary for an operation, and you believe there is a
 367 different/better way to get the same result - please file a bugreport.
 368
 369 =head1 AUTHOR
 370
 371 See L<DBIx::Class/AUTHOR> and L<DBIx::Class/CONTRIBUTORS>.
 372
 373 =head1 LICENSE
 374
 375 You may distribute this code under the same terms as Perl itself.
 376
 377 =cut