lib/DBIx/Class/Storage/DBI/Pg.pm

   1 package DBIx::Class::Storage::DBI::Pg;
   2
   3 use strict;
   4 use warnings;
   5
   6 use base qw/DBIx::Class::Storage::DBI/;
   7
   8 use Scope::Guard ();
   9 use Context::Preserve 'preserve_context';
  10 use DBIx::Class::Carp;
  11 use Try::Tiny;
  12 use namespace::clean;
  13 use DBIx::Class::Storage::DBI::Pg::Sth;
  14
  15 __PACKAGE__->sql_limit_dialect ('LimitOffset');
  16 __PACKAGE__->sql_quote_char ('"');
  17 __PACKAGE__->datetime_parser_type ('DateTime::Format::Pg');
  18 __PACKAGE__->_use_multicolumn_in (1);
  19
  20 __PACKAGE__->mk_group_accessors('simple' =>
  21                                     '_pg_cursor_number');
  22
  23 # these are package-vars to allow for evil global overrides
  24 our $DEFAULT_USE_PG_CURSORS=0;
  25 our $DEFAULT_PG_CURSORS_PAGE_SIZE=1000;
  26
  27 sub _determine_supports_insert_returning {
  28   return shift->_server_info->{normalized_dbms_version} >= 8.002
  29     ? 1
  30     : 0
  31   ;
  32 }
  33
  34 sub with_deferred_fk_checks {
  35   my ($self, $sub) = @_;
  36
  37   my $txn_scope_guard = $self->txn_scope_guard;
  38
  39   $self->_do_query('SET CONSTRAINTS ALL DEFERRED');
  40
  41   my $sg = Scope::Guard->new(sub {
  42     $self->_do_query('SET CONSTRAINTS ALL IMMEDIATE');
  43   });
  44
  45   return preserve_context { $sub->() } after => sub { $txn_scope_guard->commit };
  46 }
  47
  48 # only used when INSERT ... RETURNING is disabled
  49 sub last_insert_id {
  50   my ($self,$source,@cols) = @_;
  51
  52   my @values;
  53
  54   my $col_info = $source->columns_info(\@cols);
  55
  56   for my $col (@cols) {
  57     my $seq = ( $col_info->{$col}{sequence} ||= $self->dbh_do('_dbh_get_autoinc_seq', $source, $col) )
  58       or $self->throw_exception( sprintf(
  59         'could not determine sequence for column %s.%s, please consider adding a schema-qualified sequence to its column info',
  60           $source->name,
  61           $col,
  62       ));
  63
  64     push @values, $self->_dbh->last_insert_id(undef, undef, undef, undef, {sequence => $seq});
  65   }
  66
  67   return @values;
  68 }
  69
  70 sub _sequence_fetch {
  71   my ($self, $function, $sequence) = @_;
  72
  73   $self->throw_exception('No sequence to fetch') unless $sequence;
  74
  75   my ($val) = $self->_get_dbh->selectrow_array(
  76     sprintf ("select %s('%s')", $function, (ref $sequence eq 'SCALAR') ? $$sequence : $sequence)
  77   );
  78
  79   return $val;
  80 }
  81
  82 sub _dbh_get_autoinc_seq {
  83   my ($self, $dbh, $source, $col) = @_;
  84
  85   my $schema;
  86   my $table = $source->name;
  87
  88   # deref table name if it needs it
  89   $table = $$table
  90       if ref $table eq 'SCALAR';
  91
  92   # parse out schema name if present
  93   if( $table =~ /^(.+)\.(.+)$/ ) {
  94     ( $schema, $table ) = ( $1, $2 );
  95   }
  96
  97   # get the column default using a Postgres-specific pg_catalog query
  98   my $seq_expr = $self->_dbh_get_column_default( $dbh, $schema, $table, $col );
  99
 100   # if no default value is set on the column, or if we can't parse the
 101   # default value as a sequence, throw.
 102   unless ( defined $seq_expr and $seq_expr =~ /^nextval\(+'([^']+)'::(?:text|regclass)\)/i ) {
 103     $seq_expr = '' unless defined $seq_expr;
 104     $schema = "$schema." if defined $schema && length $schema;
 105     $self->throw_exception( sprintf (
 106       'no sequence found for %s%s.%s, check the RDBMS table definition or explicitly set the '.
 107       "'sequence' for this column in %s",
 108         $schema ? "$schema." : '',
 109         $table,
 110         $col,
 111         $source->source_name,
 112     ));
 113   }
 114
 115   return $1;
 116 }
 117
 118 # custom method for fetching column default, since column_info has a
 119 # bug with older versions of DBD::Pg
 120 sub _dbh_get_column_default {
 121   my ( $self, $dbh, $schema, $table, $col ) = @_;
 122
 123   # Build and execute a query into the pg_catalog to find the Pg
 124   # expression for the default value for this column in this table.
 125   # If the table name is schema-qualified, query using that specific
 126   # schema name.
 127
 128   # Otherwise, find the table in the standard Postgres way, using the
 129   # search path.  This is done with the pg_catalog.pg_table_is_visible
 130   # function, which returns true if a given table is 'visible',
 131   # meaning the first table of that name to be found in the search
 132   # path.
 133
 134   # I *think* we can be assured that this query will always find the
 135   # correct column according to standard Postgres semantics.
 136   #
 137   # -- rbuels
 138
 139   my $sqlmaker = $self->sql_maker;
 140   local $sqlmaker->{bindtype} = 'normal';
 141
 142   my ($where, @bind) = $sqlmaker->where ({
 143     'a.attnum' => {'>', 0},
 144     'c.relname' => $table,
 145     'a.attname' => $col,
 146     -not_bool => 'a.attisdropped',
 147     (defined $schema && length $schema)
 148       ? ( 'n.nspname' => $schema )
 149       : ( -bool => \'pg_catalog.pg_table_is_visible(c.oid)' )
 150   });
 151
 152   my ($seq_expr) = $dbh->selectrow_array(<<EOS,undef,@bind);
 153
 154 SELECT
 155   (SELECT pg_catalog.pg_get_expr(d.adbin, d.adrelid)
 156    FROM pg_catalog.pg_attrdef d
 157    WHERE d.adrelid = a.attrelid AND d.adnum = a.attnum AND a.atthasdef)
 158 FROM pg_catalog.pg_class c
 159      LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
 160      JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
 161 $where
 162
 163 EOS
 164
 165   return $seq_expr;
 166 }
 167
 168
 169 sub sqlt_type {
 170   return 'PostgreSQL';
 171 }
 172
 173 sub bind_attribute_by_data_type {
 174   my ($self,$data_type) = @_;
 175
 176   if ($self->_is_binary_lob_type($data_type)) {
 177     # this is a hot-ish codepath, use an escape flag to minimize
 178     # amount of function/method calls
 179     # additionally version.pm is cock, and memleaks on multiple
 180     # ->VERSION calls
 181     # the flag is stored in the DBD namespace, so that Class::Unload
 182     # will work (unlikely, but still)
 183     unless ($DBD::Pg::__DBIC_DBD_VERSION_CHECK_DONE__) {
 184       if ($self->_server_info->{normalized_dbms_version} >= 9.0) {
 185         try { DBD::Pg->VERSION('2.17.2'); 1 } or carp (
 186           __PACKAGE__.': BYTEA columns are known to not work on Pg >= 9.0 with DBD::Pg < 2.17.2'
 187         );
 188       }
 189       elsif (not try { DBD::Pg->VERSION('2.9.2'); 1 } ) { carp (
 190         __PACKAGE__.': DBD::Pg 2.9.2 or greater is strongly recommended for BYTEA column support'
 191       )}
 192
 193       $DBD::Pg::__DBIC_DBD_VERSION_CHECK_DONE__ = 1;
 194     }
 195
 196     return { pg_type => DBD::Pg::PG_BYTEA() };
 197   }
 198   else {
 199     return undef;
 200   }
 201 }
 202
 203 sub _exec_svp_begin {
 204     my ($self, $name) = @_;
 205
 206     $self->_dbh->pg_savepoint($name);
 207 }
 208
 209 sub _exec_svp_release {
 210     my ($self, $name) = @_;
 211
 212     $self->_dbh->pg_release($name);
 213 }
 214
 215 sub _exec_svp_rollback {
 216     my ($self, $name) = @_;
 217
 218     $self->_dbh->pg_rollback_to($name);
 219 }
 220
 221 sub deployment_statements {
 222   my $self = shift;;
 223   my ($schema, $type, $version, $dir, $sqltargs, @rest) = @_;
 224
 225   $sqltargs ||= {};
 226
 227   if (
 228     ! exists $sqltargs->{producer_args}{postgres_version}
 229       and
 230     my $dver = $self->_server_info->{normalized_dbms_version}
 231   ) {
 232     $sqltargs->{producer_args}{postgres_version} = $dver;
 233   }
 234
 235   $self->next::method($schema, $type, $version, $dir, $sqltargs, @rest);
 236 }
 237
 238 sub _populate_dbh {
 239     my ($self) = @_;
 240
 241     # cursors are per-connection, so reset the numbering
 242     $self->_pg_cursor_number(1);
 243     return $self->SUPER::_populate_dbh();
 244 }
 245
 246 sub _get_next_pg_cursor_number {
 247     my ($self) = @_;
 248
 249     my $ret=$self->_pg_cursor_number||0;
 250     $self->_pg_cursor_number($ret+1);
 251
 252     return $ret;
 253 }
 254
 255 sub __get_tweak_value {
 256     my ($self,$attrs,$slot,$default,$extra_test)=@_;
 257
 258     $extra_test||=sub{1};
 259
 260     if (   exists $attrs->{$slot}
 261         && defined $attrs->{$slot}
 262         && $extra_test->($attrs->{$slot})
 263     ) {
 264         return $attrs->{$slot};
 265     }
 266     my @info=@{$self->_dbi_connect_info};
 267     if (   @info
 268         && ref($info[-1]) eq 'HASH'
 269         && exists $info[-1]->{$slot}
 270         && defined $info[-1]->{$slot}
 271         && $extra_test->($info[-1]->{$slot})
 272     ) {
 273         return $info[-1]->{$slot};
 274     }
 275     return $default;
 276 }
 277
 278 sub _should_use_pg_cursors {
 279     my ($self,$attrs) = @_;
 280
 281     return $self->__get_tweak_value($attrs,'use_pg_cursors',$DEFAULT_USE_PG_CURSORS);
 282 }
 283
 284 sub _get_pg_cursor_page_size {
 285     my ($self,$attrs) = @_;
 286
 287     return $self->__get_tweak_value($attrs,'pg_cursors_page_size',$DEFAULT_PG_CURSORS_PAGE_SIZE,
 288                                     sub { $_[0] =~ /^\d+$/ });
 289 }
 290
 291 sub _select {
 292     my $self = shift;
 293     my ($ident, $select, $where, $attrs) = @_;
 294
 295     # ugly ugly ugly, but this is the last sub in the call chain that receives $attrs
 296     local $self->{_use_pg_cursors}=$self->_should_use_pg_cursors($attrs);
 297     local $self->{_pg_cursor_page_size}=$self->_get_pg_cursor_page_size($attrs);
 298
 299     return $self->next::method(@_);
 300 }
 301
 302 sub _dbh_sth {
 303     my ($self, $dbh, $sql) = @_;
 304
 305     if ($self->{_use_pg_cursors} && $sql =~ /^SELECT\b/i) {
 306         return DBIx::Class::Storage::DBI::Pg::Sth
 307             ->new($self,$dbh,$sql,$self->{_pg_cursor_page_size});
 308     }
 309     else { # short-circuit
 310         return $self->next::method($dbh,$sql);
 311     }
 312 }
 313
 314 1;
 315
 316 __END__
 317
 318 =head1 NAME
 319
 320 DBIx::Class::Storage::DBI::Pg - PostgreSQL-specific storage
 321
 322 =head1 SYNOPSIS
 323
 324 Automatic primary key support:
 325
 326   # In your result (table) classes
 327   use base 'DBIx::Class::Core';
 328   __PACKAGE__->set_primary_key('id');
 329
 330 Using PostgreSQL cursors on fetches:
 331
 332   my $schema = MySchemaClass->connection(
 333                    $dsn, $user, $pass,
 334                    {
 335                       use_pg_cursors => 1,
 336                       pg_cursors_page_size => 1000,
 337                    });
 338
 339   # override at ResultSet level
 340   my $rs = $schema->resultset('Something')
 341                   ->search({}, { use_pg_cursors => 0});
 342
 343 =head1 DESCRIPTION
 344
 345 This class implements autoincrements for PostgreSQL.
 346
 347 It also implements fetching data via PostgreSQL cursors, as explained
 348 in the documentation for L<DBD::Pg>.
 349
 350 =head1 CURSORS FETCHING SUPPORT
 351
 352 By default, PostgreSQL cursors are not used. You can turn them on (or
 353 off again) either via the connection attributes, or via the ResultSet
 354 attributes (the latter take precedence).
 355
 356 Fetching data using PostgreSQL cursors uses less memory, but is
 357 slightly slower. You can tune the memory / speed trade-off using the
 358 C<pg_cursors_page_size> attribute, which defines how many rows to
 359 fetch at a time (defaults to 1000).
 360
 361 =head1 POSTGRESQL SCHEMA SUPPORT
 362
 363 This driver supports multiple PostgreSQL schemas, with one caveat: for
 364 performance reasons, data about the search path, sequence names, and
 365 so forth is queried as needed and CACHED for subsequent uses.
 366
 367 For this reason, once your schema is instantiated, you should not
 368 change the PostgreSQL schema search path for that schema's database
 369 connection. If you do, Bad Things may happen.
 370
 371 You should do any necessary manipulation of the search path BEFORE
 372 instantiating your schema object, or as part of the on_connect_do
 373 option to connect(), for example:
 374
 375    my $schema = My::Schema->connect
 376                   ( $dsn,$user,$pass,
 377                     { on_connect_do =>
 378                         [ 'SET search_path TO myschema, foo, public' ],
 379                     },
 380                   );
 381
 382 =head1 AUTHORS
 383
 384 See L<DBIx::Class/CONTRIBUTORS>
 385
 386 =head1 LICENSE
 387
 388 You may distribute this code under the same terms as Perl itself.
 389
 390 =cut