lib/DBIx/Class/Schema/Loader/DBI/MSSQL.pm

   1 package DBIx::Class::Schema::Loader::DBI::MSSQL;
   2
   3 use strict;
   4 use warnings;
   5 use base 'DBIx::Class::Schema::Loader::DBI::Sybase::Common';
   6 use mro 'c3';
   7 use Try::Tiny;
   8 use List::MoreUtils 'any';
   9 use namespace::clean;
  10
  11 use DBIx::Class::Schema::Loader::Table::Sybase ();
  12
  13 our $VERSION = '0.07027';
  14
  15 =head1 NAME
  16
  17 DBIx::Class::Schema::Loader::DBI::MSSQL - DBIx::Class::Schema::Loader::DBI MSSQL Implementation.
  18
  19 =head1 DESCRIPTION
  20
  21 Base driver for Microsoft SQL Server, used by
  22 L<DBIx::Class::Schema::Loader::DBI::Sybase::Microsoft_SQL_Server> for support
  23 via L<DBD::Sybase> and
  24 L<DBIx::Class::Schema::Loader::DBI::ODBC::Microsoft_SQL_Server> for support via
  25 L<DBD::ODBC>.
  26
  27 See L<DBIx::Class::Schema::Loader> and L<DBIx::Class::Schema::Loader::Base> for
  28 usage information.
  29
  30 =head1 CASE SENSITIVITY
  31
  32 Most MSSQL databases use C<CI> (case-insensitive) collation, for this reason
  33 generated column names are lower-cased as this makes them easier to work with
  34 in L<DBIx::Class>.
  35
  36 We attempt to detect the database collation at startup for any database
  37 included in L<db_schema|DBIx::Class::Schema::Loader::Base/db_schema>, and set
  38 the column lowercasing behavior accordingly, as lower-cased column names do not
  39 work on case-sensitive databases.
  40
  41 To manually control case-sensitive mode, put:
  42
  43     preserve_case => 1|0
  44
  45 in your Loader options.
  46
  47 See L<preserve_case|DBIx::Class::Schema::Loader::Base/preserve_case>.
  48
  49 B<NOTE:> this option used to be called C<case_sensitive_collation>, but has
  50 been renamed to a more generic option.
  51
  52 =cut
  53
  54 sub _system_databases {
  55     return (qw/
  56         master model tempdb msdb
  57     /);
  58 }
  59
  60 sub _system_tables {
  61     return (qw/
  62         spt_fallback_db spt_fallback_dev spt_fallback_usg spt_monitor spt_values MSreplication_options
  63     /);
  64 }
  65
  66 sub _owners {
  67     my ($self, $db) = @_;
  68
  69     my $owners = $self->dbh->selectcol_arrayref(<<"EOF");
  70 SELECT name
  71 FROM [$db].dbo.sysusers
  72 WHERE uid <> gid
  73 EOF
  74
  75     return grep !/^(?:#|guest|INFORMATION_SCHEMA|sys)/, @$owners;
  76 }
  77
  78 sub _current_db {
  79     my $self = shift;
  80     return ($self->dbh->selectrow_array('SELECT db_name()'))[0];
  81 }
  82
  83 sub _switch_db {
  84     my ($self, $db) = @_;
  85     $self->dbh->do("use [$db]");
  86 }
  87
  88 sub _setup {
  89     my $self = shift;
  90
  91     $self->next::method(@_);
  92
  93     my $current_db = $self->_current_db;
  94
  95     if (ref $self->db_schema eq 'HASH') {
  96         if (keys %{ $self->db_schema } < 2) {
  97             my ($db) = keys %{ $self->db_schema };
  98
  99             $db ||= $current_db;
 100
 101             if ($db eq '%') {
 102                 my $owners = $self->db_schema->{$db};
 103
 104                 my $db_names = $self->dbh->selectcol_arrayref(<<'EOF');
 105 SELECT name
 106 FROM master.dbo.sysdatabases
 107 EOF
 108
 109                 my @dbs;
 110
 111                 foreach my $db_name (@$db_names) {
 112                     push @dbs, $db_name
 113                         unless any { $_ eq $db_name } $self->_system_databases;
 114                 }
 115
 116                 $self->db_schema({});
 117
 118                 DB: foreach my $db (@dbs) {
 119                     if (not ((ref $owners eq 'ARRAY' && $owners->[0] eq '%') || $owners eq '%')) {
 120                         my @owners;
 121
 122                         foreach my $owner (@$owners) {
 123                             push @owners, $owner
 124                                 if $self->dbh->selectrow_array(<<"EOF");
 125 SELECT name
 126 FROM [$db].dbo.sysusers
 127 WHERE name = @{[ $self->dbh->quote($owner) ]}
 128 EOF
 129                         }
 130
 131                         next DB unless @owners;
 132
 133                         $self->db_schema->{$db} = \@owners;
 134                     }
 135                     else {
 136                         # for post-processing below
 137                         $self->db_schema->{$db} = '%';
 138                     }
 139                 }
 140
 141                 $self->qualify_objects(1);
 142             }
 143             else {
 144                 if ($db ne $current_db) {
 145                     $self->dbh->do("USE [$db]");
 146
 147                     $self->qualify_objects(1);
 148                 }
 149             }
 150         }
 151         else {
 152             $self->qualify_objects(1);
 153         }
 154     }
 155     elsif (ref $self->db_schema eq 'ARRAY' || (not defined $self->db_schema)) {
 156         my $owners = $self->db_schema;
 157         $owners ||= [ $self->dbh->selectrow_array('SELECT user_name()') ];
 158
 159         $self->qualify_objects(1) if @$owners > 1;
 160
 161         $self->db_schema({ $current_db => $owners });
 162     }
 163
 164     foreach my $db (keys %{ $self->db_schema }) {
 165         if ($self->db_schema->{$db} eq '%') {
 166             $self->db_schema->{$db} = [ $self->_owners($db) ];
 167
 168             $self->qualify_objects(1);
 169         }
 170     }
 171
 172     if (not defined $self->preserve_case) {
 173         foreach my $db (keys %{ $self->db_schema }) {
 174             # We use the sys.databases query for the general case, and fallback to
 175             # databasepropertyex() if for some reason sys.databases is not available,
 176             # which does not work over DBD::ODBC with unixODBC+FreeTDS.
 177             #
 178             # XXX why does databasepropertyex() not work over DBD::ODBC ?
 179             #
 180             # more on collations here: http://msdn.microsoft.com/en-us/library/ms143515.aspx
 181
 182             my $current_db = $self->_current_db;
 183
 184             $self->_switch_db($db);
 185
 186             my $collation_name =
 187                    (eval { $self->dbh->selectrow_array("SELECT collation_name FROM [$db].sys.databases WHERE name = @{[ $self->dbh->quote($db) ]}") })[0]
 188                 || (eval { $self->dbh->selectrow_array("SELECT CAST(databasepropertyex(@{[ $self->dbh->quote($db) ]}, 'Collation') AS VARCHAR)") })[0];
 189
 190             $self->_switch_db($current_db);
 191
 192             if (not $collation_name) {
 193                 warn <<"EOF";
 194
 195 WARNING: MSSQL Collation detection failed for database '$db'. Defaulting to
 196 case-insensitive mode. Override the 'preserve_case' attribute in your Loader
 197 options if needed.
 198
 199 See 'preserve_case' in
 200 perldoc DBIx::Class::Schema::Loader::Base
 201 EOF
 202                 $self->preserve_case(0) unless $self->preserve_case;
 203             }
 204             else {
 205                 my $case_sensitive = $collation_name =~ /_(?:CS|BIN2?)(?:_|\z)/;
 206
 207                 if ($case_sensitive && (not $self->preserve_case)) {
 208                     $self->preserve_case(1);
 209                 }
 210                 else {
 211                     $self->preserve_case(0);
 212                 }
 213             }
 214         }
 215     }
 216 }
 217
 218 sub _tables_list {
 219     my ($self, $opts) = @_;
 220
 221     my @tables;
 222
 223     while (my ($db, $owners) = each %{ $self->db_schema }) {
 224         foreach my $owner (@$owners) {
 225             my $table_names = $self->dbh->selectcol_arrayref(<<"EOF");
 226 SELECT table_name
 227 FROM [$db].INFORMATION_SCHEMA.TABLES
 228 WHERE table_schema = @{[ $self->dbh->quote($owner) ]}
 229 EOF
 230
 231             TABLE: foreach my $table_name (@$table_names) {
 232                 next TABLE if any { $_ eq $table_name } $self->_system_tables;
 233
 234                 push @tables, DBIx::Class::Schema::Loader::Table::Sybase->new(
 235                     loader   => $self,
 236                     name     => $table_name,
 237                     database => $db,
 238                     schema   => $owner,
 239                 );
 240             }
 241         }
 242     }
 243
 244     return $self->_filter_tables(\@tables, $opts);
 245 }
 246
 247 sub _table_pk_info {
 248     my ($self, $table) = @_;
 249
 250     my $db = $table->database;
 251
 252     my $pk = $self->dbh->selectcol_arrayref(<<"EOF");
 253 SELECT kcu.column_name
 254 FROM [$db].INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc
 255 JOIN [$db].INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu
 256     ON kcu.table_name = tc.table_name
 257         AND kcu.table_schema = tc.table_schema
 258         AND kcu.constraint_name = tc.constraint_name
 259 WHERE tc.table_name = @{[ $self->dbh->quote($table->name) ]}
 260     AND tc.table_schema = @{[ $self->dbh->quote($table->schema) ]}
 261     AND tc.constraint_type = 'PRIMARY KEY'
 262 ORDER BY kcu.ordinal_position
 263 EOF
 264
 265     $pk = [ map $self->_lc($_), @$pk ];
 266
 267     return $pk;
 268 }
 269
 270 sub _table_fk_info {
 271     my ($self, $table) = @_;
 272
 273     my $db = $table->database;
 274
 275     my $sth = $self->dbh->prepare(<<"EOF");
 276 SELECT rc.constraint_name, rc.unique_constraint_schema, uk_tc.table_name, fk_kcu.column_name, uk_kcu.column_name
 277 FROM [$db].INFORMATION_SCHEMA.TABLE_CONSTRAINTS fk_tc
 278 JOIN [$db].INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS rc
 279     ON rc.constraint_name = fk_tc.constraint_name
 280         AND rc.constraint_schema = fk_tc.table_schema
 281 JOIN [$db].INFORMATION_SCHEMA.KEY_COLUMN_USAGE fk_kcu
 282     ON fk_kcu.constraint_name = fk_tc.constraint_name
 283         AND fk_kcu.table_name = fk_tc.table_name
 284         AND fk_kcu.table_schema = fk_tc.table_schema
 285 JOIN [$db].INFORMATION_SCHEMA.TABLE_CONSTRAINTS uk_tc
 286     ON uk_tc.constraint_name = rc.unique_constraint_name
 287         AND uk_tc.table_schema = rc.unique_constraint_schema
 288 JOIN [$db].INFORMATION_SCHEMA.KEY_COLUMN_USAGE uk_kcu
 289     ON uk_kcu.constraint_name = rc.unique_constraint_name
 290         AND uk_kcu.ordinal_position = fk_kcu.ordinal_position
 291         AND uk_kcu.table_name = uk_tc.table_name
 292         AND uk_kcu.table_schema = rc.unique_constraint_schema
 293 WHERE fk_tc.table_name = @{[ $self->dbh->quote($table->name) ]}
 294     AND fk_tc.table_schema = @{[ $self->dbh->quote($table->schema) ]}
 295 ORDER BY fk_kcu.ordinal_position
 296 EOF
 297
 298     $sth->execute;
 299
 300     my %rels;
 301
 302     while (my ($fk, $remote_schema, $remote_table, $col, $remote_col) = $sth->fetchrow_array) {
 303         push @{ $rels{$fk}{local_columns}  }, $self->_lc($col);
 304         push @{ $rels{$fk}{remote_columns} }, $self->_lc($remote_col);
 305
 306         $rels{$fk}{remote_table} = DBIx::Class::Schema::Loader::Table::Sybase->new(
 307             loader   => $self,
 308             name     => $remote_table,
 309             database => $db,
 310             schema   => $remote_schema,
 311         ) unless exists $rels{$fk}{remote_table};
 312     }
 313
 314     return [ values %rels ];
 315 }
 316
 317 sub _table_uniq_info {
 318     my ($self, $table) = @_;
 319
 320     my $db = $table->database;
 321
 322     my $sth = $self->dbh->prepare(<<"EOF");
 323 SELECT tc.constraint_name, kcu.column_name
 324 FROM [$db].INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc
 325 JOIN [$db].INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu
 326     ON kcu.constraint_name = tc.constraint_name
 327         AND kcu.table_name = tc.table_name
 328         AND kcu.table_schema = tc.table_schema
 329 wHERE tc.table_name = @{[ $self->dbh->quote($table->name) ]}
 330     AND tc.table_schema = @{[ $self->dbh->quote($table->schema) ]}
 331     AND tc.constraint_type = 'UNIQUE'
 332 ORDER BY kcu.ordinal_position
 333 EOF
 334
 335     $sth->execute;
 336
 337     my %uniq;
 338
 339     while (my ($constr, $col) = $sth->fetchrow_array) {
 340         push @{ $uniq{$constr} }, $self->_lc($col);
 341     }
 342
 343     return [ map [ $_ => $uniq{$_} ], keys %uniq ];
 344 }
 345
 346 sub _columns_info_for {
 347     my $self    = shift;
 348     my ($table) = @_;
 349
 350     my $db = $table->database;
 351
 352     my $result = $self->next::method(@_);
 353
 354     # SQL Server: Ancient as time itself, but still out in the wild
 355     my $is_2k = $self->schema->storage->_server_info->{normalized_dbms_version} < 9;
 356
 357     # get type info (and identity)
 358     my $rows = $self->dbh->selectall_arrayref($is_2k ? <<"EOF2K" : <<"EOF");
 359 SELECT c.column_name, c.character_maximum_length, c.data_type, c.datetime_precision, c.column_default, (sc.status & 0x80) is_identity
 360 FROM [$db].INFORMATION_SCHEMA.COLUMNS c
 361 JOIN [$db].dbo.sysusers ss ON
 362     c.table_schema = ss.name
 363 JOIN [$db].dbo.sysobjects so ON
 364     c.table_name = so.name
 365     AND so.uid = ss.uid
 366 JOIN [$db].dbo.syscolumns sc ON
 367     c.column_name = sc.name
 368     AND sc.id = so.Id
 369 WHERE c.table_schema = @{[ $self->dbh->quote($table->schema) ]}
 370     AND c.table_name = @{[ $self->dbh->quote($table->name) ]}
 371 EOF2K
 372 SELECT c.column_name, c.character_maximum_length, c.data_type, c.datetime_precision, c.column_default, sc.is_identity
 373 FROM [$db].INFORMATION_SCHEMA.COLUMNS c
 374 JOIN [$db].sys.schemas ss ON
 375     c.table_schema = ss.name
 376 JOIN [$db].sys.objects so ON
 377       c.table_name   = so.name
 378     AND so.schema_id = ss.schema_id
 379 JOIN [$db].sys.columns sc ON
 380     c.column_name = sc.name
 381     AND sc.object_id = so.object_id
 382 WHERE c.table_schema = @{[ $self->dbh->quote($table->schema) ]}
 383     AND c.table_name = @{[ $self->dbh->quote($table->name) ]}
 384 EOF
 385
 386     foreach my $row (@$rows) {
 387         my ($col, $char_max_length, $data_type, $datetime_precision, $default, $is_identity) = @$row;
 388         $col = lc $col unless $self->preserve_case;
 389         my $info = $result->{$col} || next;
 390
 391         $info->{data_type} = $data_type;
 392
 393         if (defined $char_max_length) {
 394             $info->{size} = $char_max_length;
 395             $info->{size} = 0 if $char_max_length < 0;
 396         }
 397
 398         if ($is_identity) {
 399             $info->{is_auto_increment} = 1;
 400             $info->{data_type} =~ s/\s*identity//i;
 401             delete $info->{size};
 402         }
 403
 404         # fix types
 405         if ($data_type eq 'int') {
 406             $info->{data_type} = 'integer';
 407         }
 408         elsif ($data_type eq 'timestamp') {
 409             $info->{inflate_datetime} = 0;
 410         }
 411         elsif ($data_type =~ /^(?:numeric|decimal)\z/) {
 412             if (ref($info->{size}) && $info->{size}[0] == 18 && $info->{size}[1] == 0) {
 413                 delete $info->{size};
 414             }
 415         }
 416         elsif ($data_type eq 'float') {
 417             $info->{data_type} = 'double precision';
 418             delete $info->{size};
 419         }
 420         elsif ($data_type =~ /^(?:small)?datetime\z/) {
 421             # fixup for DBD::Sybase
 422             if ($info->{default_value} && $info->{default_value} eq '3') {
 423                 delete $info->{default_value};
 424             }
 425         }
 426         elsif ($data_type =~ /^(?:datetime(?:2|offset)|time)\z/) {
 427             $info->{size} = $datetime_precision;
 428
 429             delete $info->{size} if $info->{size} == 7;
 430         }
 431         elsif ($data_type eq 'varchar'   && $info->{size} == 0) {
 432             $info->{data_type} = 'text';
 433             delete $info->{size};
 434         }
 435         elsif ($data_type eq 'nvarchar'  && $info->{size} == 0) {
 436             $info->{data_type} = 'ntext';
 437             delete $info->{size};
 438         }
 439         elsif ($data_type eq 'varbinary' && $info->{size} == 0) {
 440             $info->{data_type} = 'image';
 441             delete $info->{size};
 442         }
 443
 444         if ($data_type !~ /^(?:n?char|n?varchar|binary|varbinary|numeric|decimal|float|datetime(?:2|offset)|time)\z/) {
 445             delete $info->{size};
 446         }
 447
 448         if (defined $default) {
 449             # strip parens
 450             $default =~ s/^\( (.*) \)\z/$1/x;
 451
 452             # Literal strings are in ''s, numbers are in ()s (in some versions of
 453             # MSSQL, in others they are unquoted) everything else is a function.
 454             $info->{default_value} =
 455                 $default =~ /^['(] (.*) [)']\z/x ? $1 :
 456                     $default =~ /^\d/ ? $default : \$default;
 457
 458             if ((eval { lc ${ $info->{default_value} } }||'') eq 'getdate()') {
 459                 ${ $info->{default_value} } = 'current_timestamp';
 460
 461                 my $getdate = 'getdate()';
 462                 $info->{original}{default_value} = \$getdate;
 463             }
 464         }
 465     }
 466
 467     return $result;
 468 }
 469
 470 =head1 SEE ALSO
 471
 472 L<DBIx::Class::Schema::Loader::DBI::Sybase::Microsoft_SQL_Server>,
 473 L<DBIx::Class::Schema::Loader::DBI::ODBC::Microsoft_SQL_Server>,
 474 L<DBIx::Class::Schema::Loader>, L<DBIx::Class::Schema::Loader::Base>,
 475 L<DBIx::Class::Schema::Loader::DBI>
 476
 477 =head1 AUTHOR
 478
 479 See L<DBIx::Class::Schema::Loader/AUTHOR> and L<DBIx::Class::Schema::Loader/CONTRIBUTORS>.
 480
 481 =head1 LICENSE
 482
 483 This library is free software; you can redistribute it and/or modify it under
 484 the same terms as Perl itself.
 485
 486 =cut
 487
 488 1;
 489 # vim:et sts=4 sw=4 tw=0: