File Coverage

blib/lib/DBD/Crate.pm

Criterion	Covered	Total	%
statement	69	202	34.1
branch	0	54	0.0
condition	0	25	0.0
subroutine	23	41	56.1
pod	0	3	0.0
total	92	325	28.3

package DBD::Crate;
use strict;
use DBI;
use HTTP::Tiny;
use JSON::MaybeXS;
use vars qw($VERSION $REVISION);
use vars qw($err $errstr $state $drh);
$VERSION = "0.0.3";

$err     = 0;
$errstr  = "";
$state   = "";
$drh     = undef;
my $methods_are_installed = 0;
my ($HTTP, $JSON);

sub driver {
    return $drh if $drh;
    my ($class, $attr) = @_;
    $class .= "::dr";
    $drh = DBI::_new_drh($class, {
        'Name'        => 'Crate',
        'Version'     => $VERSION,
        'Err'         => \$err,
        'Errstr'      => \$errstr,
        'State'       => \$state,
        'Attribution' => 'DBD::Crate by Mamod Mehyar',
        'AutoCommit'  => 1
    }) or return undef;
    return $drh;
}

sub http { $HTTP }
sub json { $JSON }

#====================================================================
# DBD::Crate::dr
#====================================================================
package DBD::Crate::dr; {
    use strict;
    use DBI qw(:sql_types);
    use vars qw($imp_data_size);
    use Carp qw(carp croak);
    use Data::Dumper;
    use DBI;

    $imp_data_size = 0;

    sub connect {
        my ($drh, $dburl, $user, $pass, $attr) = @_;
        my $UTF8 = defined $attr->{utf8} ?
                        $attr->{utf8} : 1;

        $JSON = JSON::MaybeXS->new({ utf8 => $UTF8 });
        $HTTP = HTTP::Tiny->new( keep_alive => 1 );

        my @addresses = ($dburl);
        my @addr;
        if ($dburl =~ s/^\[(.*?)\]$/$1/){
            @addresses = split ',', $dburl;
        }

        foreach my $addr (@addresses){
            $addr =~ s/\s+//;
            if (!$addr){
                $addr = 'http://localhost:4200';
            }

            if ($addr !~ /^http/){
                $addr = 'http://' . $addr;
            }

            if ($user || $pass){
                my $auth = ($user || '') . ':' . ($pass || '');
                $addr =~ s/^(http(?:.)?:\/\/)(.*?)/$1$auth\@$2/;
            }
            push @addr, $addr;
        }

        my ($t, $dbh) = DBI::_new_dbh($drh, {
            'Name'   => \@addr
        });

        return $dbh;
    }

    sub data_sources { return "Cratedb" }
    sub disconnect_all { 1 }
};

#====================================================================
# DBD::Crate::db
#====================================================================
package DBD::Crate::db; {
    use strict;
    use base qw(DBD::_::db);
    use vars qw($imp_data_size);
    use Data::Dumper;
    use DBI;
    use Digest::SHA1  qw(sha1_hex);

    $imp_data_size = 0;

    sub prepare {
        my ($dbh, $statement, @attr) = @_;
        my $sth = DBI::_new_sth($dbh, {
            'Statement'      => $statement,
            'ConnectionHOST' => $dbh->{Name}
        });
        return $sth;
    }

    #=============================================================
    # blob methods
    #=============================================================
    sub crate_blob_insert {
        my ($dbh, $table, $digest, $content) = @_;

        if (!$content){
            $content = $digest;
            $digest = sha1_hex($content);
        }

        my $path = "/_blobs/$table/" . $digest;
        my $sth = DBI::_new_sth($dbh, {
            'REQUEST_PATH'   => $path,
            'REQUEST_METHOD' => 'PUT',
            'Statement'      => $content,
            'ConnectionHOST' => $dbh->{Name},
            'BLOB'           => 1,
            'DIGEST'         => $digest
        });

        return $sth->execute();
    };

    sub crate_blob_get {
        my ($dbh, $table, $digest) = @_;

        if (!$digest){
            $dbh->set_err(-1, "BLOB sha1 digest required");
            return;
        }

        $digest ||= '';

        my $path = "/_blobs/$table/" . $digest;
        my $sth = DBI::_new_sth($dbh, {
            'REQUEST_PATH'   => $path,
            'REQUEST_METHOD' => 'GET',
            'Statement'      => '',
            'ConnectionHOST' => $dbh->{Name},
            'BLOB'           => 1,
            'DIGEST'         => $digest
        });
        return $sth->execute();
    }

    sub crate_blob_delete {
        my ($dbh, $table, $digest) = @_;

        if (!$digest){
            $dbh->set_err(-1, "BLOB sha1 digest required");
            return;
        }

        my $path = "/_blobs/$table/" . $digest;
        my $sth = DBI::_new_sth($dbh, {
            'REQUEST_PATH'   => $path,
            'REQUEST_METHOD' => 'DELETE',
            'Statement'      => '',
            'ConnectionHOST' => $dbh->{Name},
            'BLOB'           => 1,
            'DIGEST'         => $digest
        });

        return $sth->execute();
    }

    #=============================================================
    # table info methods
    #=============================================================
    #return columns information of provided table
    sub crate_table_columns {
        my ($dbh, $table) = @_;
        my $sth = $dbh->prepare(qq~
            select column_name, data_type, ordinal_position
            from information_schema.columns
            where schema_name = 'doc'
            AND table_name = ?
        ~);

        return $dbh->selectall_arrayref( $sth,
            { Slice => {} },
            $table);
    }

    #list all tables with information
    sub crate_tables_list {
        my $dbh = shift;
        my $schema = shift || 'doc';
        my $sth = $dbh->prepare(qq~
            select number_of_replicas, partitioned_by, blobs_path, schema_name,
                   table_name, number_of_shards, clustered_by
                   from information_schema.tables
                   where schema_name = ?
        ~);

        return $dbh->selectall_arrayref( $sth,
            { Slice => {} },
            $schema);
    }

    #get table info
    sub crate_table_info {
        my $dbh   = shift;
        my $table = shift;
        my $sth = $dbh->prepare(qq~
            select number_of_replicas, partitioned_by, blobs_path, schema_name,
                   table_name, number_of_shards, clustered_by
                   from information_schema.tables
                   where schema_name = 'doc'
                   AND table_name = ?
        ~);

        return $dbh->selectrow_hashref( $sth,
            undef,
            $table);
    }

    #==================================================
    # These should be removed once we get crate name
    # space registered with DBI
    #==================================================
    *DBI::db::crate_blob_insert   = \&crate_blob_insert;
    *DBI::db::crate_blob_get      = \&crate_blob_get;
    *DBI::db::crate_blob_delete   = \&crate_blob_delete;
    *DBI::db::crate_table_columns = \&crate_table_columns;
    *DBI::db::crate_tables_list   = \&crate_tables_list;
    *DBI::db::crate_table_info    = \&crate_table_info;
};

#====================================================================
# DBD::Crate::st
#====================================================================
package DBD::Crate::st; {
    use strict;
    use base qw(DBD::_::st);
    use vars qw($imp_data_size);
    use Data::Dumper;
    use DBI;

    $imp_data_size = 0;

    sub _fetch_data {
        my $sth       = shift;
        my $statement = shift;
        my $method    = shift || 'POST';
        my $path      = shift || '/_sql';

        my @hosts = @{ $sth->{ConnectionHOST} };

        TRYAGAIN :
        my $host = shift @hosts;
        my $ret = DBD::Crate::http->request($method, $host . $path, {
            content => $statement,
            # headers => {'Content-Type' => 'application/json'}
        });

        if ( $ret->{status} == 599 && scalar @hosts){
            my $i = 0;
            for (@{ $sth->{ConnectionHOST} }){
                #put failing address to the end of hosts
                #this will work only on persistant environments
                if ( $host eq $_ ){
                    splice @{ $sth->{ConnectionHOST} }, $i, 1;
                    push @{ $sth->{ConnectionHOST} }, $_;
                }
                $i++;
            }
            goto TRYAGAIN;
        }

        if (!$ret->{success}){
            my $olderr = $@;
            my $data = eval { DBD::Crate::json->decode($ret->{content}) };
            $@ = $olderr;

            my $error = ref $data eq 'HASH' && ref $data->{error} ? $data->{error} : {
                code    => ref $data ? $data->{status} : $ret->{status},
                message => ref $data ? $data->{error}  : ($ret->{content} || $ret->{reason})
            };

            $sth->set_err($error->{code}, $error->{message});
            return;
        }

        return $ret;
    }

    sub execute {
        my $sth = shift;
        my $statement = $sth->{Statement};
        my $ret;
        if ($sth->{BLOB}){
            if (!$sth->{DIGEST}){
                $sth->set_err(-1, "BLOB sha1 digest required");
                return;
            }

            $ret = _fetch_data($sth, $statement,
                                $sth->{REQUEST_METHOD},
                                $sth->{REQUEST_PATH}) or return;
        } else {
            my $hash = {stmt => $statement };
            if (@_){ $hash->{args} = \@_; }
            my $json = DBD::Crate::json->encode($hash);
            $ret = _fetch_data($sth, $json, 'POST', '/_sql') or return;
        }

        $sth->{'driver_raw_data'} = $ret->{content};
        if ($sth->{BLOB}){
            if ($ret->{status} == 201){ # put success
                return $sth->{DIGEST};
            } elsif ($ret->{status} == 200){ #get success
                return $ret->{content};
            } elsif ($ret->{status} == 204){ #delete success
                return 1;
            }
        }

        my $olderr = $@;
        my $data = eval { DBD::Crate::json->decode($ret->{content}) };
        if (!$data){
            my $error = $@;
            $@ = $olderr;
            $sth->set_err(-1, $error);
            return;
        }


        $sth->{'driver_data'} = $data->{rows};
        $sth->{'driver_rows'} =  $data->{rowcount};

        $sth->{'NAME'} = $data->{cols};
        $sth->STORE('NUM_OF_FIELDS', scalar @{ $data->{cols} });
        return $data->{rowcount} || '0E0';
    }

    *fetch = \&fetchrow_arrayref;
    sub fetchrow_arrayref {
        my $sth = shift;
        my $data = $sth->FETCH('driver_data');
        my $row = shift @$data or return;
        return $sth->_set_fbav($row);
    }

    sub raw {
        my $sth = shift;
        my $data = $sth->FETCH('driver_raw_data');
        return $data;
    }

    *DBI::st::raw = \&raw;

    #Nothing to close, crate is stateless
    sub close {}
};


1;

__END__