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