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