File Coverage

blib/lib/PDL/IO/DBI.pm

Criterion	Covered	Total	%
statement	52	268	19.4
branch	3	188	1.6
condition	0	50	0.0
subroutine	17	24	70.8
pod	2	2	100.0
total	74	532	13.9

package PDL::IO::DBI;

use strict;
use warnings;

use Exporter 'import';
our @EXPORT_OK   = qw(rdbi1D rdbi2D);
our %EXPORT_TAGS = (all => \@EXPORT_OK);

our $VERSION = '0.011';

use Config;
use constant NO64BITINT => ($Config{ivsize} < 8) ? 1 : 0;
use constant NODATETIME => eval { require PDL::DateTime; require Time::Moment; 1 } ? 0 : 1;
use constant DEBUG      => $ENV{PDL_IO_DBI_DEBUG} ? 1 : 0;

use PDL;
use DBI;
use Time::Moment;

use Carp;
$Carp::Internal{ (__PACKAGE__) }++;

my %pck = (
  byte     => "C",
  short    => "s",
  ushort   => "S",
  long     => "l",
  longlong => "q",
  float    => "f",
  double   => "d",
);

my %tmap = (
  DBI::SQL_TINYINT   => byte,        # -6
  DBI::SQL_BIGINT    => longlong,    # -5
  DBI::SQL_NUMERIC   => double,      #  2
  DBI::SQL_DECIMAL   => double,      #  3
  DBI::SQL_INTEGER   => long,        #  4
  DBI::SQL_SMALLINT  => short,       #  5
  DBI::SQL_FLOAT     => double,      #  6
  DBI::SQL_REAL      => float,       #  7
  DBI::SQL_DOUBLE    => double,      #  8
  DBI::SQL_DATETIME  => '_dt_',      #  9 == DBI::SQL_DATE
 #DBI::SQL_INTERVAL  => longlong,    # 10 == DBI::SQL_TIME
  DBI::SQL_TIMESTAMP => '_dt_',      # 11
  DBI::SQL_BOOLEAN   => byte,        # 16
  DBI::SQL_TYPE_DATE => '_dt_',      # 91
 #DBI::SQL_TYPE_TIME                 # 92
  DBI::SQL_TYPE_TIMESTAMP => '_dt_', # 93
 #DBI::SQL_TYPE_TIME_WITH_TIMEZONE   # 94
  DBI::SQL_TYPE_TIMESTAMP_WITH_TIMEZONE => '_dt_', # 95
 #DBI::SQL_INTERVAL_YEAR                 101
 #DBI::SQL_INTERVAL_MONTH                102
 #DBI::SQL_INTERVAL_DAY                  103
 #DBI::SQL_INTERVAL_HOUR                 104
 #DBI::SQL_INTERVAL_MINUTE               105
 #DBI::SQL_INTERVAL_SECOND               106
 #DBI::SQL_INTERVAL_YEAR_TO_MONTH        107
 #DBI::SQL_INTERVAL_DAY_TO_HOUR          108
 #DBI::SQL_INTERVAL_DAY_TO_MINUTE        109
 #DBI::SQL_INTERVAL_DAY_TO_SECOND        110
 #DBI::SQL_INTERVAL_HOUR_TO_MINUTE       111
 #DBI::SQL_INTERVAL_HOUR_TO_SECOND       112
 #DBI::SQL_INTERVAL_MINUTE_TO_SECOND     113
  ################## DBD::SQLite uses text values instead of numerical constants corresponding to DBI::SQL_*
  'BIGINT'           => longlong, # 8 bytes, -9223372036854775808 .. 9223372036854775807
  'INT8'             => longlong, # 8 bytes
  'INTEGER'          => long,     # 4 bytes, -2147483648 .. 2147483647
  'INT'              => long,     # 4 bytes
  'INT4'             => long,     # 4 bytes
  'MEDIUMINT'        => long,     # 3 bytes, -8388608 .. 8388607
  'SMALLINT'         => short,    # 2 bytes, -32768 .. 32767
  'INT2'             => short,    # 2 bytes
  'TINYINT'          => byte,     # 1 byte, MySQL: -128 .. 127, MSSQL+Pg: 0 to 255
  'REAL'             => float,    # 4 bytes
  'FLOAT'            => double,   # 8 bytes
  'NUMERIC'          => double,
  'DECIMAL'          => double,
  'DOUBLE'           => double,
  'DOUBLE PRECISION' => double,
  'BOOLEAN'          => byte,
  'SMALLSERIAL'      => short,    # 2 bytes, 1 to 32767
  'SERIAL'           => long,     # 4 bytes, 1 to 2147483647
  'BIGSERIAL'        => longlong, # 8 bytes, 1 to 9223372036854775807
  'DATETIME'         => '_dt_',
  'DATE'             => '_dt_',
  'TIMESTAMP'        => '_dt_',
);

# https://www.sqlite.org/datatype3.html
# http://dev.mysql.com/doc/refman/5.7/en/integer-types.html
# http://www.postgresql.org/docs/9.3/static/datatype-numeric.html
# http://msdn.microsoft.com/en-us/library/ff848794.aspx

sub _dt_to_double {
  my $str = shift;
  $str = $str."T00Z" if $str =~ /^\d\d\d\d-\d\d-\d\d$/;
  $str = $str."Z"    if $str !~ /(UTC|GMT|Z|\+)/;
  my $t = eval { Time::Moment->from_string($str, lenient=>1) } or die "INVALID DATETIME: $str"; #XXX-FIXME PDL::dt2ll / dt2dbl ?
  return $t->epoch * 1.0 + $t->millisecond / 1_000;
}

sub _dt_to_longlong {
  my $str = shift;
  $str = $str."T00Z" if $str =~ /^\d\d\d\d-\d\d-\d\d$/;
  $str = $str."Z"    if $str !~ /(UTC|GMT|Z|\+)/;
  my $t = eval { Time::Moment->from_string($str, lenient=>1) } or die "INVALID DATETIME: $str"; #XXX-FIXME PDL::dt2ll
  return $t->epoch*1000000 + $t->microsecond;
}

sub rdbi1D {
  my ($dbh, $sql, $bind_values, $O) = _proc_args(@_);

  # reuse_sth (if defined) is a scalar reference to a statement handle to be reused
  my $reuse_sth = $O->{reuse_sth};
  my $sth;
  if ($reuse_sth && $$reuse_sth) {
    $sth = $$reuse_sth;
  }
  else {
    $sth = $dbh->prepare($sql) or croak "FATAL: prepare failed: " . $dbh->errstr;
    $sth->execute(@$bind_values)  or croak "FATAL: execute failed: " . $sth->errstr;
    $$reuse_sth = $sth if $reuse_sth;
  }

  my ($c_type, $c_pack, $c_sizeof, $c_pdl, $c_bad, $c_dataref, $c_idx, $c_convert, $allocated, $cols) = _init_1D($sth->{TYPE}, $O);
  warn "Initial size: '$allocated'\n" if $O->{debug};
  my $null2bad = $O->{null2bad};
  my $processed = 0;
  my $headerline = $sth->{NAME_lc};

  warn "Fetching data (type=", join(',', @$c_type), ") ...\n" if $O->{debug};
  while (my $data = $sth->fetchall_arrayref(undef, $O->{fetch_chunk})) { # limiting MaxRows
    my $rows = scalar @$data;
    if ($rows > 0) {
      $processed += $rows;
      if ($allocated < $processed) {
        $allocated += $O->{reshape_inc};
        warn "Reshape to: '$allocated'\n" if $O->{debug};
        for (0..$cols-1) {
          $c_pdl->[$_]->reshape($allocated);
          $c_dataref->[$_] = $c_pdl->[$_]->get_dataref;
        }
      }
      if ($null2bad) {
        for my $tmp (@$data) {
          for (0..$cols-1) {
            unless (defined $tmp->[$_]) {
              $tmp->[$_] = $c_bad->[$_];
              $c_pdl->[$_]->badflag(1);
            }
          }
        }
      }
      if (scalar @$c_convert > 0) {
        for my $c (0..$cols-1) {
          if (ref $c_convert->[$c] eq 'CODE') {
            for my $r (0..$rows-1) {
              $data->[$r]->[$c] = $c_convert->[$c]->($data->[$r]->[$c]);
            }
          }
        }
      }
      for my $ci (0..$cols-1) {
        my $bytes = '';
        {
          no warnings 'pack'; # intentionally disable all pack related warnings
          no warnings 'numeric'; # disable: Argument ??? isn't numeric in pack
          no warnings 'uninitialized'; # disable: Use of uninitialized value in pack
          $bytes .= pack($c_pack->[$ci], $data->[$_][$ci]) for(0..$rows-1);
        }
        my $len = length $bytes;
        my $expected_len = $c_sizeof->[$ci] * $rows;
        croak "FATAL: len mismatch $len != $expected_len" if $len != $expected_len;
        substr(${$c_dataref->[$ci]}, $c_idx->[$ci], $len) = $bytes;
        $c_idx->[$ci] += $expected_len;
      }
    }
    last if $reuse_sth;
  }
  croak "FATAL: DB fetch failed: " . $sth->errstr if $sth->err;

  if ($processed != $allocated) {
    warn "Reshape to: '$processed' (final)\n" if $O->{debug};
    $c_pdl->[$_]->reshape($processed) for (0..$cols-1);
  }
  $c_pdl->[$_]->upd_data for (0..$cols-1);
  if (ref $headerline eq 'ARRAY') {
    for (0..$cols-1) {
      $c_pdl->[$_]->hdr->{col_name} = $headerline->[$_] if $headerline->[$_] && $headerline->[$_] ne '';
    };
  }

  if ($processed == 0) {
    if ($reuse_sth) {
      # signal to callers that all chunks have been fetched
      $$reuse_sth = undef;
    }
    else {
      warn "rdbi1D: no data\n";
    }
  }

  return @$c_pdl;
}

sub rdbi2D {
  my ($dbh, $sql, $bind_values, $O) = _proc_args(@_);

  croak 'FATAL: reuse_sth not supported yet for rdbi2D' if $O->{reuse_sth};
  my $sth = $dbh->prepare($sql) or croak "FATAL: prepare failed: " . $dbh->errstr;
  $sth->execute(@$bind_values) or croak "FATAL: execute failed: " . $sth->errstr;

  my ($c_type, $c_pack, $c_sizeof, $c_pdl, $c_bad, $c_dataref, $c_convert, $allocated, $cols) = _init_2D($sth->{TYPE}, $O);
  warn "Initial size: '$allocated'\n" if $O->{debug};
  my $null2bad = $O->{null2bad};
  my $processed = 0;
  my $c_idx = 0;
  my $pck = "$c_pack\[$cols\]";

  warn "Fetching data (type=$c_type) ...\n" if $O->{debug};
  while (my $data = $sth->fetchall_arrayref(undef, $O->{fetch_chunk})) { # limiting MaxRows
    my $rows = scalar @$data;
    if ($rows > 0) {
      $processed += $rows;
      if ($allocated < $processed) {
        $allocated += $O->{reshape_inc};
        warn "Reshape to: '$allocated'\n" if $O->{debug};
        $c_pdl->reshape($cols, $allocated);
        $c_dataref = $c_pdl->get_dataref;
      }
      my $bytes = '';
      if ($null2bad) {
        for my $tmp (@$data) {
          for (@$tmp) {
            unless (defined $_) {
              $_ = $c_bad;
              $c_pdl->badflag(1);
            }
          }
        }
      }
      if (scalar @$c_convert > 0) {
        for my $c (0..$cols-1) {
          if (ref $c_convert->[$c] eq 'CODE') {
            for my $r (0..$rows-1) {
              $data->[$r]->[$c] = $c_convert->[$c]->($data->[$r]->[$c]);
            }
          }
        }
      }
      {
          no warnings 'pack'; # intentionally disable all pack related warnings
          no warnings 'numeric'; # disable: Argument ??? isn't numeric in pack
          no warnings 'uninitialized'; # disable: Use of uninitialized value in pack
        $bytes .= pack($pck, @$_) for (@$data);
      }
      my $len = length $bytes;
      my $expected_len = $c_sizeof * $cols * $rows;
      croak "FATAL: len mismatch $len != $expected_len" if $len != $expected_len;
      substr($$c_dataref, $c_idx, $len) = $bytes;
      $c_idx += $len;
    }
  }
  croak "FATAL: DB fetch failed: " . $sth->errstr if $sth->err;
  if ($processed != $allocated) {
    warn "Reshape to: '$processed' (final)\n" if $O->{debug};
    $c_pdl->reshape($cols, $processed); # allocate the exact size
  }
  $c_pdl->upd_data;

  warn "rdbi2D: no data\n" unless $processed > 0;

  return $c_pdl->transpose;
}

sub _proc_args {
  my $options = ref $_[-1] eq 'HASH' ? pop : {};
  my ($dsn_or_dbh, $sql, $bind_values) = @_;

  croak "FATAL: no SQL query"  unless $sql;
  croak "FATAL: no DBH or DSN" unless defined $dsn_or_dbh;
  my $reuse_sth = $options->{reuse_sth};
  if ($reuse_sth) {
    croak "FATAL: reuse_sth must either be false, a reference to a false value, or a reference to a statement handle"
      if $$reuse_sth && !eval { $$reuse_sth->isa('DBI::st') };
  }
  my $O = { %$options }; # make a copy

  # handle defaults for optional parameters
  $O->{fetch_chunk} =  8_000 unless defined $O->{fetch_chunk};
  my $alloc = $reuse_sth ? $O->{fetch_chunk} : 80_000;
  $O->{reshape_inc} = $alloc unless defined $O->{reshape_inc};
  $O->{type}        = 'auto' unless defined $O->{type};
  $O->{debug}       = DEBUG  unless defined $O->{debug};

  # reshape_inc cannot be lower than fetch_chunk
  $O->{reshape_inc} = $O->{fetch_chunk} if $O->{reshape_inc} < $O->{fetch_chunk};

  $bind_values = [] unless ref $bind_values eq 'ARRAY';

  # launch db query
  my $dbh = ref $dsn_or_dbh ? $dsn_or_dbh : DBI->connect($dsn_or_dbh) or croak "FATAL: connect failed: " . $DBI::errstr;

  return ($dbh, $sql, $bind_values, $O);
}

sub _init_1D {
  my ($sql_types, $O) = @_;

  croak "FATAL: no columns" unless ref $sql_types eq 'ARRAY';
  my $cols = scalar @$sql_types;
  croak "FATAL: no columns" unless $cols > 0;

  my @c_type;
  my @c_pack;
  my @c_sizeof;
  my @c_pdl;
  my @c_bad;
  my @c_dataref;
  my @c_idx;
  my @c_convert;

  if (ref $O->{type} eq 'ARRAY') {
    @c_type = @{$O->{type}};
  }
  else {
    $c_type[$_] = $O->{type} for (0..$cols-1);
  }
  for (0..$cols-1) { $c_type[$_] = 'auto' if !$c_type[$_] }

  my @detected_type = map { $sql_types->[$_] ? $tmap{$sql_types->[$_]} : undef } (0..$cols-1);
  if ($O->{debug}) {
    $detected_type[$_] or warn "column $_ has unknown type '$sql_types->[$_]' gonna use Double\n" for (0..$cols-1);
  }
  my $allocated = $O->{reshape_inc};

  my @c_dt;
  for (0..$cols-1) {
    if ($detected_type[$_] && $detected_type[$_] eq '_dt_') {
      if (!NODATETIME && ($c_type[$_] eq 'auto' || $c_type[$_] eq 'datetime')) {
        croak "PDL::DateTime not installed" if NODATETIME;
        $c_convert[$_] = \&_dt_to_longlong;
        $c_type[$_] = longlong;
        $c_dt[$_] = 'datetime';
      }
      elsif ($c_type[$_] eq longlong) {
        $c_convert[$_] = \&_dt_to_longlong;
      }
      else {
        $c_convert[$_] = \&_dt_to_double;
        $c_type[$_] = double;
      }
    }
    $c_type[$_] = $detected_type[$_] if !defined $c_type[$_] || $c_type[$_] eq 'auto';
    $c_type[$_] = double if !$c_type[$_];
    $c_pack[$_] = $pck{$c_type[$_]};
    croak "FATAL: your perl does not support 64bitint (avoid using type longlong)" if $c_pack[$_] eq 'q' && NO64BITINT;
    croak "FATAL: invalid type '$c_type[$_]' for column $_" if !$c_pack[$_];
    $c_sizeof[$_] = length pack($c_pack[$_], 1);
    $c_pdl[$_] = $c_dt[$_] ? PDL::DateTime->new(zeroes(longlong, $allocated)) : zeroes($c_type[$_], $allocated);
    $c_dataref[$_] = $c_pdl[$_]->get_dataref;
    $c_bad[$_] = $c_pdl[$_]->badvalue;
    $c_idx[$_] = 0;
    my $big = PDL::Core::howbig($c_pdl[$_]->get_datatype);
    croak "FATAL: column $_ mismatch (type=$c_type[$_], sizeof=$c_sizeof[$_], big=$big)" if $big != $c_sizeof[$_];
  }

  return (\@c_type, \@c_pack, \@c_sizeof, \@c_pdl, \@c_bad, \@c_dataref, \@c_idx, \@c_convert, $allocated, $cols);
}

sub _init_2D {
  my ($sql_types, $O) = @_;

  croak "FATAL: no columns" unless ref $sql_types eq 'ARRAY';
  my $cols = scalar @$sql_types;
  croak "FATAL: no columns" unless $cols > 0;

  my $c_type = $O->{type};
  if (!$c_type || $c_type eq 'auto') {
    # try to guess the best type
    my @detected_type = map { $sql_types->[$_] ? $tmap{$sql_types->[$_]} : undef } (0..$cols-1);
    if ($O->{debug}) {
      $detected_type[$_] or warn "column $_ has unknown type '$sql_types->[$_]' gonna use Double\n" for (0..$cols-1);
    }
    for (0..$#detected_type) {
      # DATETIME is auto-detected as double
      $detected_type[$_] = 'double' if $detected_type[$_] && $detected_type[$_] eq '_dt_';
      my $dt = $detected_type[$_] || 'double';
      $c_type = double    if $dt eq double;
      $c_type = float     if $dt eq float    && $c_type ne double;
      $c_type = longlong  if $dt eq longlong && $c_type !~ /^(double|float)$/;
      $c_type = long      if $dt eq long     && $c_type !~ /^(double|float|longlong)$/;
      $c_type = short     if $dt eq short    && $c_type !~ /^(double|float|longlong|long)$/;
      $c_type = byte      if $dt eq byte     && $c_type !~ /^(double|float|longlong|long|short)$/;
    }
    croak "FATAL: type detection failed" if !$c_type;
  }
  my $c_pack = $pck{$c_type};
  croak "FATAL: your perl does not support 64bitint (avoid using type longlong)" if $c_pack eq 'q' && NO64BITINT;
  croak "FATAL: invalid type '$c_type' for column $_" if !$c_pack;

  my @c_convert = ();
  for (0..$cols-1) {
    my $t = $tmap{$sql_types->[$_]} || '';
    if ($t eq '_dt_') {
        $c_convert[$_] = ($c_type eq 'longlong') ? \&_dt_to_longlong : \&_dt_to_double;
    }
  }

  my $allocated = $O->{reshape_inc};
  my $c_sizeof = length pack($c_pack, 1);
  my $c_pdl = zeroes($c_type, $cols, $allocated);
  my $c_dataref = $c_pdl->get_dataref;
  my $c_bad = $c_pdl->badvalue;

  my $howbig = PDL::Core::howbig($c_pdl->get_datatype);
  croak "FATAL: column $_ size mismatch (type=$c_type, sizeof=$c_sizeof, howbig=$howbig)" unless  $howbig == $c_sizeof;

  return ($c_type, $c_pack, $c_sizeof, $c_pdl, $c_bad, $c_dataref, \@c_convert, $allocated, $cols);
}

1;

__END__