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