monetdb.pm

来自「这个是内存数据库的客户端」· PM 代码 · 共 884 行 · 第 1/2 页

package DBD::monetdb;

use strict;
use sigtrap;
use DBI();
use MonetDB::CLI();

our $VERSION = '0.09';
our $drh = undef;

require DBD::monetdb::GetInfo;
require DBD::monetdb::TypeInfo;


sub driver {
    return $drh if $drh;

    my ($class, $attr) = @_;

    $drh = DBI::_new_drh($class .'::dr', {
        Name        => 'monetdb',
        Version     => $VERSION,
        Attribution => 'DBD::monetdb by Martin Kersten, Arjan Scherpenisse and Steffen Goeldner',
    });
}


sub CLONE {
    undef $drh;
}



package DBD::monetdb::dr;

$DBD::monetdb::dr::imp_data_size = 0;


sub connect {
    my ($drh, $dsn, $user, $password, $attr) = @_;

    my %dsn;
    for ( split /;|:/, $dsn ||'') {
        if ( my ( $k, $v ) = /(.*?)=(.*)/) {
            $k = 'host'     if $k eq 'hostname';
            $k = 'database' if $k eq 'dbname' || $k eq 'db';
            $dsn{$k} = $v;
            next;
        }
        for my $k ( qw(host port database language) ) {
            $dsn{$k} = $_, last unless defined $dsn{$k};
        }
    }
    my $lang  = $dsn{language} || 'sql';
    my $host  = $dsn{host} || 'localhost';
    my $port  = $dsn{port} || 50000;
    $user     ||= 'monetdb';
    $password ||= 'monetdb';

    my $cxn = eval { MonetDB::CLI->connect($host, $port, $user, $password, $lang) };
    return $drh->set_err(-1, $@) if $@;

    my ($outer, $dbh) = DBI::_new_dbh($drh, { Name => $dsn });

    $dbh->STORE('Active', 1 );

    $dbh->{monetdb_connection} = $cxn;
    $dbh->{monetdb_language} = $lang;

    return $outer;
}


sub data_sources {
    return ('dbi:monetdb:');
}



package DBD::monetdb::db;

$DBD::monetdb::db::imp_data_size = 0;


sub ping {
    my ($dbh) = @_;

    my $statement = $dbh->{monetdb_language} eq 'sql' ? 'select 7' : 'print(7);';
    my $rv = $dbh->selectrow_array($statement) || 0;
    $dbh->set_err(undef, undef);
    $rv == 7 ? 1 : 0;
}


sub quote {
    my ($dbh, $value, $type) = @_;

    return $dbh->{monetdb_language} eq 'sql' ? 'NULL' : 'nil'
        unless defined $value;

    for ($value) {
      s/	/\\t/g;
      s/\\/\\\\/g;
      s/\n/\\n/g;
      s/\r/\\r/g;
      s/"/\\"/g;
      s/'/''/g;
    }

    $type ||= DBI::SQL_VARCHAR();

    my $prefix = $DBD::monetdb::TypeInfo::prefixes{$type} || '';
    my $suffix = $DBD::monetdb::TypeInfo::suffixes{$type} || '';

    if ( $dbh->{monetdb_language} ne 'sql') {
        $prefix = q(") if $prefix eq q(');
        $suffix = q(") if $suffix eq q(');
    }
    return $prefix . $value . $suffix;
}


sub _count_param {
    my @statement = split //, shift;
    my $num = 0;

    while (defined(my $c = shift @statement)) {
        if ($c eq '"' || $c eq "'") {
            my $end = $c;
            while (defined(my $c = shift @statement)) {
                last if $c eq $end;
                @statement = splice @statement, 2 if $c eq '\\';
            }
        }
        elsif ($c eq '?') {
            $num++;
        }
    }
    return $num;
}


sub prepare {
    my ($dbh, $statement, $attr) = @_;

    my $cxn = $dbh->{monetdb_connection};
    my $hdl = eval { $cxn->new_handle };
    return $dbh->set_err(-1, $@) if $@;

    my ($outer, $sth) = DBI::_new_sth($dbh, { Statement => $statement });

    $sth->STORE('NUM_OF_PARAMS', _count_param($statement));

    $sth->{monetdb_hdl} = $hdl;
    $sth->{monetdb_params} = [];
    $sth->{monetdb_types} = [];
    $sth->{monetdb_rows} = -1;

    return $outer;
}


sub commit {
    my($dbh) = @_;

    if ($dbh->FETCH('AutoCommit')) {
        warn 'Commit ineffective while AutoCommit is on' if $dbh->FETCH('Warn');
        return 0;
    }
    if ($dbh->{monetdb_language} eq 'sql') {
        return $dbh->do('commit')
            && $dbh->do('start transaction');
    }
    else {
        return $dbh->do('commit();');
    }
}


sub rollback {
    my($dbh) = @_;

    if ($dbh->FETCH('AutoCommit')) {
        warn 'Rollback ineffective while AutoCommit is on' if $dbh->FETCH('Warn');
        return 0;
    }
    if ($dbh->{monetdb_language} eq 'sql') {
        return $dbh->do('rollback')
            && $dbh->do('start transaction');
    }
    else {
        return $dbh->do('abort();');
    }
}


*get_info = \&DBD::monetdb::GetInfo::get_info;


sub monetdb_catalog_info {
    my($dbh) = @_;
    my $sql = <<'SQL';
select cast( null as varchar( 128 ) ) as table_cat
     , cast( null as varchar( 128 ) ) as table_schem
     , cast( null as varchar( 128 ) ) as table_name
     , cast( null as varchar( 254 ) ) as table_type
     , cast( null as varchar( 254 ) ) as remarks
 where 0 = 1
 order by table_cat
SQL
    my $sth = $dbh->prepare($sql) or return;
    $sth->execute or return;
    return $sth;
}


sub monetdb_schema_info {
    my($dbh) = @_;
    my $sql = <<'SQL';
select cast( null as varchar( 128 ) ) as table_cat
     , "name"                         as table_schem
     , cast( null as varchar( 128 ) ) as table_name
     , cast( null as varchar( 254 ) ) as table_type
     , cast( null as varchar( 254 ) ) as remarks
  from sys."schemas"
 order by table_schem
SQL
    my $sth = $dbh->prepare($sql) or return;
    $sth->execute or return;
    return $sth;
}


my $ttp = {
 'TABLE'           => 't."istable" = true  and t."system" = false and t."temporary" = 0'
,'SYSTEM TABLE'    => 't."istable" = true  and t."system" = true  and t."temporary" = 0'
,'LOCAL TEMPORARY' => 't."istable" = true  and t."system" = false and t."temporary" = 1'
,'VIEW'            => 't."istable" = false                                             '
};


sub monetdb_tabletype_info {
    my($dbh) = @_;
    my $sql = <<"SQL";
select distinct
       cast( null as varchar( 128 ) ) as table_cat
     , cast( null as varchar( 128 ) ) as table_schem
     , cast( null as varchar( 128 ) ) as table_name
     , case
         when $ttp->{'TABLE'          } then cast('TABLE'               as varchar( 254 ) )
         when $ttp->{'SYSTEM TABLE'   } then cast('SYSTEM TABLE'        as varchar( 254 ) )
         when $ttp->{'LOCAL TEMPORARY'} then cast('LOCAL TEMPORARY'     as varchar( 254 ) )
         when $ttp->{'VIEW'           } then cast('VIEW'                as varchar( 254 ) )
         else                                cast('INTERNAL TABLE TYPE' as varchar( 254 ) )
       end                            as table_type
     , cast( null as varchar( 254 ) ) as remarks
  from sys."tables" t
 order by table_type
SQL
    my $sth = $dbh->prepare($sql) or return;
    $sth->execute or return;
    return $sth;
}


sub monetdb_table_info {
    my($dbh, $c, $s, $t, $tt) = @_;
    my $sql = <<"SQL";
select cast( null as varchar( 128 ) ) as table_cat
     , s."name"                       as table_schem
     , t."name"                       as table_name
     , case
         when $ttp->{'TABLE'          } then cast('TABLE'               as varchar( 254 ) )
         when $ttp->{'SYSTEM TABLE'   } then cast('SYSTEM TABLE'        as varchar( 254 ) )
         when $ttp->{'LOCAL TEMPORARY'} then cast('LOCAL TEMPORARY'     as varchar( 254 ) )
         when $ttp->{'VIEW'           } then cast('VIEW'                as varchar( 254 ) )
         else                                cast('INTERNAL TABLE TYPE' as varchar( 254 ) )
       end                            as table_type
     , cast( null as varchar( 254 ) ) as remarks
  from sys."schemas" s
     , sys."tables"  t
 where t."schema_id" = s."id"
SQL
    my @bv = ();
    $sql .= qq(   and s."name"   like ?\n), push @bv, $s if $s;
    $sql .= qq(   and t."name"   like ?\n), push @bv, $t if $t;
    if ( @$tt ) {
        $sql .= "   and ( 1 = 0\n";
        for ( @$tt ) {
            my $p = $ttp->{uc $_};
            $sql .= "      or $p\n" if $p;
        }
        $sql .= "       )\n";
    }
    $sql .=   " order by table_type, table_schem, table_name\n";
    my $sth = $dbh->prepare($sql) or return;
    $sth->execute(@bv) or return;
    $dbh->set_err(0,"Catalog parameter '$c' ignored") if defined $c;
    return $sth;
}


sub table_info {
    my($dbh, $c, $s, $t, $tt) = @_;

    if ( defined $c && defined $s && defined $t ) {
        if    ( $c eq '%' && $s eq ''  && $t eq '') {
            return monetdb_catalog_info($dbh);
        }
        elsif ( $c eq ''  && $s eq '%' && $t eq '') {
            return monetdb_schema_info($dbh);
        }
        elsif ( $c eq ''  && $s eq ''  && $t eq '' && defined $tt && $tt eq '%') {
            return monetdb_tabletype_info($dbh);
        }
    }
    my @tt;
    if ( defined $tt ) {
        @tt = split /,/, $tt;
        s/^\s*'?//, s/'?\s*$// for @tt;
    }
    return monetdb_table_info($dbh, $c, $s, $t, \@tt);
}


sub column_info {
    my($dbh, $catalog, $schema, $table, $column) = @_;
    my $sql = <<'SQL';
select cast( null            as varchar( 128 ) ) as table_cat
     , s."name"                                  as table_schem
     , t."name"                                  as table_name
     , c."name"                                  as column_name
     , cast( 0               as smallint       ) as data_type          -- ...
     , c."type"                                  as type_name          -- TODO
     , cast( c."type_digits" as integer        ) as column_size        -- TODO
     , cast( null            as integer        ) as buffer_length      -- TODO
     , cast( c."type_scale"  as smallint       ) as decimal_digits     -- TODO
     , cast( null            as smallint       ) as num_prec_radix     -- TODO
     , case c."null"
         when false then cast( 0 as smallint )  -- SQL_NO_NULLS
         when true  then cast( 1 as smallint )  -- SQL_NULLABLE
       end                                       as nullable
     , cast( null            as varchar( 254 ) ) as remarks
     , c."default"                               as column_def
     , cast( 0               as smallint       ) as sql_data_type      -- ...
     , cast( null            as smallint       ) as sql_datetime_sub   -- ...
     , cast( null            as integer        ) as char_octet_length  -- TODO
     , cast( c."number" + 1  as integer        ) as ordinal_position
     , case c."null"
         when false then cast('NO'  as varchar( 254 ) )
         when true  then cast('YES' as varchar( 254 ) )
       end                                       as is_nullable
  from sys."schemas" s
     , sys."tables"  t
     , sys."columns" c
 where t."schema_id" = s."id"
   and c."table_id"  = t."id"
SQL
    my @bv = ();
    $sql .= qq(   and s."name"   like ?\n), push @bv, $schema if $schema;
    $sql .= qq(   and t."name"   like ?\n), push @bv, $table  if $table;
    $sql .= qq(   and c."name"   like ?\n), push @bv, $column if $column;
    $sql .=   " order by table_cat, table_schem, table_name, ordinal_position\n";
    my $sth = $dbh->prepare($sql) or return;
    $sth->execute(@bv) or return;
    $dbh->set_err(0,"Catalog parameter '$catalog' ignored") if defined $catalog;
    my $rows;
    while ( my $row = $sth->fetch ) {
        $row->[ 4] = $DBD::monetdb::TypeInfo::typeinfo{$row->[5]}->[ 1];
        $row->[13] = $DBD::monetdb::TypeInfo::typeinfo{$row->[5]}->[15];
        $row->[14] = $DBD::monetdb::TypeInfo::typeinfo{$row->[5]}->[16];
        push @$rows, [ @$row ];
    }
    return DBI->connect('dbi:Sponge:','','', { RaiseError => 1 } )->prepare(
        $sth->{Statement},
        { rows => $rows, NAME => $sth->{NAME}, TYPE => $sth->{TYPE} }
    );
}


sub primary_key_info {
    my($dbh, $catalog, $schema, $table) = @_;
    return $dbh->set_err(-1,'Undefined schema','HY009') unless defined $schema;
    return $dbh->set_err(-1,'Undefined table' ,'HY009') unless defined $table;
    my $sql = <<'SQL';
select cast( null        as varchar( 128 ) ) as table_cat
     , s."name"                              as table_schem
     , t."name"                              as table_name
     , c."column"                            as column_name
     , cast( c."nr" + 1  as smallint       ) as key_seq
     , k."name"                              as pk_name
  from sys."schemas"     s
     , sys."tables"      t
     , sys."keys"        k
     , sys."keycolumns"  c
 where t."schema_id"   = s."id"
   and k."table_id"    = t."id"
   and c."id"          = k."id"
   and s."name"        = ?
   and t."name"        = ?
   and k."type"        = 0
 order by table_schem, table_name, key_seq
SQL
    my $sth = $dbh->prepare($sql) or return;
    $sth->execute($schema, $table) or return;
    $dbh->set_err(0,"Catalog parameter '$catalog' ignored") if defined $catalog;
    return $sth;
}


sub foreign_key_info {
    my($dbh, $c1, $s1, $t1, $c2, $s2, $t2) = @_;
    my $sql = <<'SQL';
select cast( null         as varchar( 128 ) ) as uk_table_cat
     , uks."name"                             as uk_table_schem
     , ukt."name"                             as uk_table_name
     , ukc."column"                           as uk_column_name
     , cast( null         as varchar( 128 ) ) as fk_table_cat
     , fks."name"                             as fk_table_schem
     , fkt."name"                             as fk_table_name
     , fkc."column"                           as fk_column_name
     , cast( fkc."nr" + 1 as smallint       ) as ordinal_position
     , cast( 3            as smallint       ) as update_rule    -- SQL_NO_ACTION
     , cast( 3            as smallint       ) as delete_rule    -- SQL_NO_ACTION
     , fkk."name"                             as fk_name
     , ukk."name"                             as uk_name
     , cast( 7            as smallint       ) as deferability   -- SQL_NOT_DEFERRABLE
     , case  ukk."type"
         when 0 then cast('PRIMARY'   as varchar( 7 ) )
         when 1 then cast('UNIQUE'    as varchar( 7 ) )
         else        cast( ukk."type" as varchar( 7 ) )
       end                                    as unique_or_primary
  from sys."schemas"    uks
     , sys."tables"     ukt
     , sys."keys"       ukk
     , sys."keycolumns" ukc
     , sys."schemas"    fks
     , sys."tables"     fkt
     , sys."keys"       fkk
     , sys."keycolumns" fkc
 where ukt."schema_id"  = uks."id"
   and ukk."table_id"   = ukt."id"