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