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

   1 package DBIx::Class::Storage::DBI::Pg::Sth;
   2 use strict;
   3 use warnings;
   4 use base 'Class::Accessor::Grouped';
   5
   6 __PACKAGE__->mk_group_accessors('simple' =>
   7                                     'storage',
   8                                     'cursor_id', 'cursor_sql',
   9                                     'cursor_created',
  10                                     'cursor_sth', 'fetch_sth',
  11                                     'page_size',
  12                             );
  13
  14 sub new {
  15     my ($class, $storage, $dbh, $sql, $page_size) = @_;
  16
  17     if ($sql =~ /^SELECT\b/i) {
  18         my $self=bless {},$class;
  19         $self->storage($storage);
  20
  21         my $csr_id=$self->_cursor_name_from_number(
  22             $storage->_get_next_pg_cursor_number()
  23         );
  24         my $hold= ($sql =~ /\bFOR\s+UPDATE\s*\z/i) ? '' : 'WITH HOLD';
  25         $self->cursor_sql("DECLARE $csr_id CURSOR $hold FOR $sql");
  26         $self->cursor_id($csr_id);
  27         $self->cursor_sth(undef);
  28         $self->cursor_created(0);
  29         $self->page_size($page_size);
  30         return $self;
  31     }
  32     else {
  33         die "Can only be used for SELECTs";
  34     }
  35 }
  36
  37 sub _cursor_name_from_number {
  38     return 'dbic_pg_cursor_'.$_[1];
  39 }
  40
  41 sub _prepare_cursor_sth {
  42     my ($self)=@_;
  43
  44     return if $self->cursor_sth;
  45
  46     $self->cursor_sth($self->storage->sth($self->cursor_sql));
  47 }
  48
  49 sub _cleanup_sth {
  50     my ($self)=@_;
  51
  52     if ($self->fetch_sth) {
  53         $self->fetch_sth->finish();
  54         $self->fetch_sth(undef);
  55     }
  56     if ($self->cursor_sth) {
  57         $self->cursor_sth->finish();
  58         $self->cursor_sth(undef);
  59         $self->storage->dbh->do('CLOSE '.$self->cursor_id);
  60     }
  61 }
  62
  63 sub DESTROY {
  64     my ($self) = @_;
  65
  66     eval { $self->_cleanup_sth };
  67
  68     return;
  69 }
  70
  71 sub bind_param {
  72     my ($self,@bind_args)=@_;
  73
  74     $self->_prepare_cursor_sth;
  75
  76     return $self->cursor_sth->bind_param(@bind_args);
  77 }
  78
  79 sub execute {
  80     my ($self,@bind_values)=@_;
  81
  82     $self->_prepare_cursor_sth;
  83
  84     my $ret=$self->cursor_sth->execute(@bind_values);
  85     $self->cursor_created(1) if $ret;
  86     return $ret;
  87 }
  88
  89 # bind_param_array & execute_array not used for SELECT statements, so
  90 # we'll ignore them
  91
  92 sub errstr {
  93     my ($self)=@_;
  94
  95     return $self->cursor_sth->errstr;
  96 }
  97
  98 sub finish {
  99     my ($self)=@_;
 100
 101     $self->fetch_sth->finish if $self->fetch_sth;
 102     return $self->cursor_sth->finish if $self->cursor_sth;
 103     return 1;
 104 }
 105
 106 sub _check_cursor_end {
 107     my ($self) = @_;
 108
 109     if ($self->fetch_sth->rows == 0) {
 110         $self->_cleanup_sth;
 111         return 1;
 112     }
 113     return;
 114 }
 115
 116 sub _run_fetch_sth {
 117     my ($self)=@_;
 118
 119     if (!$self->cursor_created) {
 120         $self->execute();
 121     }
 122
 123     $self->fetch_sth->finish if $self->fetch_sth;
 124     $self->fetch_sth($self->storage->sth(
 125         sprintf 'fetch %d from %s',
 126         $self->page_size,
 127         $self->cursor_id
 128     ));
 129     $self->fetch_sth->execute;
 130 }
 131
 132 sub fetchrow_array {
 133     my ($self) = @_;
 134
 135     $self->_run_fetch_sth unless $self->fetch_sth;
 136     return if $self->_check_cursor_end;
 137
 138     my @row = $self->fetch_sth->fetchrow_array;
 139     if (!@row) {
 140         $self->_run_fetch_sth;
 141         return if $self->_check_cursor_end;
 142
 143         @row = $self->fetch_sth->fetchrow_array;
 144     }
 145     return @row;
 146 }
 147
 148 sub fetchall_arrayref {
 149     my ($self,$slice,$max_rows) = @_;
 150
 151     my $ret=[];
 152     $self->_run_fetch_sth unless $self->fetch_sth;
 153     return if $self->_check_cursor_end;
 154
 155     while (1) {
 156         my $batch=$self->fetch_sth->fetchall_arrayref($slice,$max_rows);
 157
 158         push @$ret,@$batch;
 159
 160         if (defined($max_rows) && $max_rows >=0) {
 161             $max_rows -= @$batch;
 162             last if $max_rows <=0;
 163         }
 164
 165         last if @$batch ==0;
 166
 167         $self->_run_fetch_sth;
 168         last if $self->_check_cursor_end;
 169     }
 170
 171     return $ret;
 172 }
 173
 174 1;