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::SQLAHacks::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 = 2 ** 32;
 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_maker {
 200   my $self = shift;
 201
 202   unless ($self->_sql_maker) {
 203     unless ($self->{_sql_maker_opts}{limit_dialect}) {
 204       my $have_rno = 0;
 205
 206       if (exists $self->_server_info->{normalized_dbms_version}) {
 207         $have_rno = 1 if $self->_server_info->{normalized_dbms_version} >= 9;
 208       }
 209       else {
 210         # User is connecting via DBD::Sybase and has no permission to run
 211         # stored procedures like xp_msver, or version detection failed for some
 212         # other reason.
 213         # So, we use a query to check if RNO is implemented.
 214         try {
 215           $self->_get_dbh->selectrow_array('SELECT row_number() OVER (ORDER BY rand())');
 216           $have_rno = 1;
 217         };
 218       }
 219
 220       $self->{_sql_maker_opts} = {
 221         limit_dialect => ($have_rno ? 'RowNumberOver' : 'Top'),
 222         %{$self->{_sql_maker_opts}||{}}
 223       };
 224     }
 225
 226     my $maker = $self->next::method (@_);
 227   }
 228
 229   return $self->_sql_maker;
 230 }
 231
 232 sub _ping {
 233   my $self = shift;
 234
 235   my $dbh = $self->_dbh or return 0;
 236
 237   local $dbh->{RaiseError} = 1;
 238   local $dbh->{PrintError} = 0;
 239
 240   return try {
 241     $dbh->do('select 1');
 242     1;
 243   } catch {
 244     0;
 245   };
 246 }
 247
 248 package # hide from PAUSE
 249   DBIx::Class::Storage::DBI::MSSQL::DateTime::Format;
 250
 251 my $datetime_format      = '%Y-%m-%d %H:%M:%S.%3N'; # %F %T
 252 my $smalldatetime_format = '%Y-%m-%d %H:%M:%S';
 253
 254 my ($datetime_parser, $smalldatetime_parser);
 255
 256 sub parse_datetime {
 257   shift;
 258   require DateTime::Format::Strptime;
 259   $datetime_parser ||= DateTime::Format::Strptime->new(
 260     pattern  => $datetime_format,
 261     on_error => 'croak',
 262   );
 263   return $datetime_parser->parse_datetime(shift);
 264 }
 265
 266 sub format_datetime {
 267   shift;
 268   require DateTime::Format::Strptime;
 269   $datetime_parser ||= DateTime::Format::Strptime->new(
 270     pattern  => $datetime_format,
 271     on_error => 'croak',
 272   );
 273   return $datetime_parser->format_datetime(shift);
 274 }
 275
 276 sub parse_smalldatetime {
 277   shift;
 278   require DateTime::Format::Strptime;
 279   $smalldatetime_parser ||= DateTime::Format::Strptime->new(
 280     pattern  => $smalldatetime_format,
 281     on_error => 'croak',
 282   );
 283   return $smalldatetime_parser->parse_datetime(shift);
 284 }
 285
 286 sub format_smalldatetime {
 287   shift;
 288   require DateTime::Format::Strptime;
 289   $smalldatetime_parser ||= DateTime::Format::Strptime->new(
 290     pattern  => $smalldatetime_format,
 291     on_error => 'croak',
 292   );
 293   return $smalldatetime_parser->format_datetime(shift);
 294 }
 295
 296 1;
 297
 298 =head1 NAME
 299
 300 DBIx::Class::Storage::DBI::MSSQL - Base Class for Microsoft SQL Server support
 301 in DBIx::Class
 302
 303 =head1 SYNOPSIS
 304
 305 This is the base class for Microsoft SQL Server support, used by
 306 L<DBIx::Class::Storage::DBI::ODBC::Microsoft_SQL_Server> and
 307 L<DBIx::Class::Storage::DBI::Sybase::Microsoft_SQL_Server>.
 308
 309 =head1 IMPLEMENTATION NOTES
 310
 311 =head2 IDENTITY information
 312
 313 Microsoft SQL Server supports three methods of retrieving the IDENTITY
 314 value for inserted row: IDENT_CURRENT, @@IDENTITY, and SCOPE_IDENTITY().
 315 SCOPE_IDENTITY is used here because it is the safest.  However, it must
 316 be called is the same execute statement, not just the same connection.
 317
 318 So, this implementation appends a SELECT SCOPE_IDENTITY() statement
 319 onto each INSERT to accommodate that requirement.
 320
 321 C<SELECT @@IDENTITY> can also be used by issuing:
 322
 323   $self->_identity_method('@@identity');
 324
 325 it will only be used if SCOPE_IDENTITY() fails.
 326
 327 This is more dangerous, as inserting into a table with an on insert trigger that
 328 inserts into another table with an identity will give erroneous results on
 329 recent versions of SQL Server.
 330
 331 =head2 identity insert
 332
 333 Be aware that we have tried to make things as simple as possible for our users.
 334 For MSSQL that means that when a user tries to create a row, while supplying an
 335 explicit value for an autoincrementing column, we will try to issue the
 336 appropriate database call to make this possible, namely C<SET IDENTITY_INSERT
 337 $table_name ON>. Unfortunately this operation in MSSQL requires the
 338 C<db_ddladmin> privilege, which is normally not included in the standard
 339 write-permissions.
 340
 341 =head2 Ordered Subselects
 342
 343 If you attempted the following query (among many others) in Microsoft SQL
 344 Server
 345
 346  $rs->search ({}, {
 347   prefetch => 'relation',
 348   rows => 2,
 349   offset => 3,
 350  });
 351
 352 You may be surprised to receive an exception. The reason for this is a quirk
 353 in the MSSQL engine itself, and sadly doesn't have a sensible workaround due
 354 to the way DBIC is built. DBIC can do truly wonderful things with the aid of
 355 subselects, and does so automatically when necessary. The list of situations
 356 when a subselect is necessary is long and still changes often, so it can not
 357 be exhaustively enumerated here. The general rule of thumb is a joined
 358 L<has_many|DBIx::Class::Relationship/has_many> relationship with limit/group
 359 applied to the left part of the join.
 360
 361 In its "pursuit of standards" Microsft SQL Server goes to great lengths to
 362 forbid the use of ordered subselects. This breaks a very useful group of
 363 searches like "Give me things number 4 to 6 (ordered by name), and prefetch
 364 all their relations, no matter how many". While there is a hack which fools
 365 the syntax checker, the optimizer may B<still elect to break the subselect>.
 366 Testing has determined that while such breakage does occur (the test suite
 367 contains an explicit test which demonstrates the problem), it is relative
 368 rare. The benefits of ordered subselects are on the other hand too great to be
 369 outright disabled for MSSQL.
 370
 371 Thus compromise between usability and perfection is the MSSQL-specific
 372 L<resultset attribute|DBIx::Class::ResultSet/ATTRIBUTES> C<unsafe_subselect_ok>.
 373 It is deliberately not possible to set this on the Storage level, as the user
 374 should inspect (and preferably regression-test) the return of every such
 375 ResultSet individually. The example above would work if written like:
 376
 377  $rs->search ({}, {
 378   unsafe_subselect_ok => 1,
 379   prefetch => 'relation',
 380   rows => 2,
 381   offset => 3,
 382  });
 383
 384 If it is possible to rewrite the search() in a way that will avoid the need
 385 for this flag - you are urged to do so. If DBIC internals insist that an
 386 ordered subselect is necessary for an operation, and you believe there is a
 387 different/better way to get the same result - please file a bugreport.
 388
 389 =head1 AUTHOR
 390
 391 See L<DBIx::Class/AUTHOR> and L<DBIx::Class/CONTRIBUTORS>.
 392
 393 =head1 LICENSE
 394
 395 You may distribute this code under the same terms as Perl itself.
 396
 397 =cut