FS/FS/Cursor.pm

   1 package FS::Cursor;
   2
   3 use strict;
   4 use vars qw($DEBUG $buffer);
   5 use FS::Record;
   6 use FS::UID qw(myconnect driver_name);
   7 use Scalar::Util qw(refaddr blessed);
   8
   9 $DEBUG = 2;
  10
  11 # this might become a parameter at some point, but right now, you can
  12 # "local $FS::Cursor::buffer = X;"
  13 $buffer = 200;
  14
  15 =head1 NAME
  16
  17 FS::Cursor - Iterator for querying large data sets
  18
  19 =head1 SYNOPSIS
  20
  21 use FS::Cursor;
  22
  23 my $search = FS::Cursor->new('table', { field => 'value' ... });
  24 while ( my $row = $search->fetch ) {
  25 ...
  26 }
  27
  28 =head1 CLASS METHODS
  29
  30 =over 4
  31
  32 =item new ARGUMENTS [, DBH ]
  33
  34 Constructs a cursored search.  Accepts all the same arguments as qsearch,
  35 and returns an FS::Cursor object to fetch the rows one at a time.
  36
  37 DBH may be a database handle; if so, the cursor will be created on that
  38 connection and have all of its transaction state. Otherwise a new connection
  39 will be opened for the cursor.
  40
  41 =cut
  42
  43 sub new {
  44   my $class = shift;
  45   my $dbh;
  46   if ( blessed($_[-1]) and $_[-1]->isa('DBI::db') ) {
  47     $dbh = pop;
  48   }
  49   my $q = FS::Record::_query(@_); # builds the statement and parameter list
  50
  51   my $self = {
  52     query => $q,
  53     class => 'FS::' . ($q->{table} || 'Record'),
  54     buffer => [],
  55     position => 0, # for mysql
  56   };
  57   bless $self, $class;
  58
  59   # the class of record object to return
  60   $self->{class} = "FS::".($q->{table} || 'Record');
  61
  62   # save for later, so forked children will not destroy me when they exit
  63   $self->{pid} = $$;
  64
  65   $self->{id} = sprintf('cursor%08x', refaddr($self));
  66
  67   my $statement;
  68   if ( driver_name() eq 'Pg' ) {
  69     if (!$dbh) {
  70       $dbh = myconnect();
  71       $self->{autoclean} = 1;
  72     }
  73     $self->{dbh} = $dbh;
  74     $statement = "DECLARE ".$self->{id}." CURSOR FOR ".$q->{statement};
  75   } elsif ( driver_name() eq 'mysql' ) {
  76     # build a cursor from scratch
  77     #
  78     #
  79     # there are problems doing it this way, and we don't have time to resolve
  80     # them all right now...
  81     #$statement = "CREATE TEMPORARY TABLE $self->{id}
  82     #  (rownum INT AUTO_INCREMENT, PRIMARY KEY (rownum))
  83     #  $q->{statement}";
  84
  85     # one of those problems is locking, so keep everything on the main session
  86     $self->{dbh} = $dbh = FS::UID::dbh();
  87     $statement = $q->{statement};
  88   }
  89
  90   my $sth = $dbh->prepare($statement)
  91     or die $dbh->errstr;
  92   my $bind = 1;
  93   foreach my $value ( @{ $q->{value} } ) {
  94     my $bind_type = shift @{ $q->{bind_type} };
  95     $sth->bind_param($bind++, $value, $bind_type );
  96   }
  97
  98   $sth->execute or die $sth->errstr;
  99
 100   if ( driver_name() eq 'Pg' ) {
 101     $self->{fetch} = $dbh->prepare("FETCH FORWARD $buffer FROM ".$self->{id});
 102   } elsif ( driver_name() eq 'mysql' ) {
 103     # make sure we're not holding any locks on the tables mentioned
 104     # in the query
 105     #$dbh->commit if driver_name() eq 'mysql';
 106     #$self->{fetch} = $dbh->prepare("SELECT * FROM $self->{id} ORDER BY rownum LIMIT ?, $buffer");
 107
 108     # instead, fetch all the rows at once
 109     $self->{buffer} = $sth->fetchall_arrayref( {} );
 110   }
 111
 112   $self;
 113 }
 114
 115 =back
 116
 117 =head1 METHODS
 118
 119 =over 4
 120
 121 =item fetch
 122
 123 Fetch the next row from the search results.
 124
 125 =cut
 126
 127 sub fetch {
 128   # might be a little more efficient to do a FETCH NEXT 1000 or something
 129   # and buffer them locally, but the semantics are simpler this way
 130   my $self = shift;
 131   if (@{ $self->{buffer} } == 0) {
 132     my $rows = $self->refill;
 133     return undef if !$rows;
 134   }
 135   $self->{class}->new(shift @{ $self->{buffer} });
 136 }
 137
 138 sub refill {
 139   my $self = shift;
 140   if (driver_name() eq 'Pg') {
 141     my $sth = $self->{fetch};
 142     $sth->bind_param(1, $self->{position}) if driver_name() eq 'mysql';
 143     $sth->execute or die $sth->errstr;
 144     my $result = $self->{fetch}->fetchall_arrayref( {} );
 145     $self->{buffer} = $result;
 146     $self->{position} += $sth->rows;
 147     scalar @$result;
 148   } # mysql can't be refilled, since everything is buffered from the start
 149 }
 150
 151 sub DESTROY {
 152   my $self = shift;
 153   return if driver_name() eq 'mysql';
 154
 155   return unless $self->{pid} eq $$;
 156   $self->{dbh}->do('CLOSE '. $self->{id})
 157     or die $self->{dbh}->errstr; # clean-up the cursor in Pg
 158   if ($self->{autoclean}) {
 159     # the dbh was created just for this cursor, so it has no transaction
 160     # state that we care about
 161     $self->{dbh}->rollback;
 162   }
 163 }
 164
 165 =back
 166
 167 =head1 TO DO
 168
 169 Replace all uses of qsearch with this.
 170
 171 =head1 BUGS
 172
 173 Still doesn't really support MySQL, but it pretends it does, by simply
 174 running the query and returning records one at a time.
 175
 176 =head1 SEE ALSO
 177
 178 L<FS::Record>
 179
 180 =cut
 181
 182 1;