File Coverage

blib/lib/Alzabo/Runtime/Schema.pm

Criterion	Covered	Total	%
statement	30	239	12.5
branch	0	106	0.0
condition	0	21	0.0
subroutine	10	37	27.0
pod	22	23	95.6
total	62	426	14.5

package Alzabo::Runtime::Schema;

use strict;
use vars qw($VERSION);

use Alzabo::Exceptions ( abbr => [ qw( logic_exception params_exception ) ] );
use Alzabo::Runtime;
use Alzabo::Utils;

use Params::Validate qw( :all );
Params::Validate::validation_options( on_fail => sub { params_exception join '', @_ } );

use base qw(Alzabo::Schema);

$VERSION = 2.0;

1;

sub load_from_file
{
    my $class = shift;

    my $self = $class->_load_from_file(@_);

    $self->prefetch_all_but_blobs;

    return $self;
}

sub _schema_file_type
{
    return 'runtime';
}

sub user
{
    my $self = shift;

    return $self->{user};
}

sub password
{
    my $self = shift;

    return $self->{password};
}

sub host
{
    my $self = shift;

    return $self->{host};
}

sub port
{
    my $self = shift;

    return $self->{port};
}

sub referential_integrity
{
    my $self = shift;

    return defined $self->{maintain_integrity} ? $self->{maintain_integrity} : 0;
}

sub set_db_schema_name
{
    my $self = shift;

    $self->{db_schema_name} = shift;
}

sub set_user
{
    my $self = shift;

    $self->{user} = shift;
}

sub set_password
{
    my $self = shift;

    $self->{password} = shift;
}

sub set_host
{
    my $self = shift;

    $self->{host} = shift;
}

sub set_port
{
    my $self = shift;

    $self->{port} = shift;
}

sub set_referential_integrity
{
    my $self = shift;
    my $val = shift;

    $self->{maintain_integrity} = $val if defined $val;
}

sub set_quote_identifiers
{
    my $self = shift;
    my $val = shift;

    $self->{quote_identifiers} = $val if defined $val;
}

sub connect
{
    my $self = shift;

    my %p;
    $p{user} = $self->user if defined $self->user;
    $p{password} = $self->password if defined $self->password;
    $p{host} = $self->host if defined $self->host;
    $p{port} = $self->port if defined $self->port;
    $self->driver->connect( %p, @_ );

#    $self->set_referential_integrity( ! $self->driver->supports_referential_integrity );
}

sub disconnect
{
    my $self = shift;

    $self->driver->disconnect;
}

sub one_row
{
    # could be replaced with something potentially more efficient
    return shift->join(@_)->next;
}

use constant JOIN_SPEC => { join => { type => ARRAYREF | OBJECT,
                                      optional => 1 },
                            tables => { type => ARRAYREF | OBJECT,
                                        optional => 1 },
                            select => { type => ARRAYREF | OBJECT,
                                        optional => 1 },
                            where => { type => ARRAYREF,
                                       optional => 1 },
                            order_by => { type => ARRAYREF | HASHREF | OBJECT,
                                          optional => 1 },
                            limit => { type => SCALAR | ARRAYREF,
                                       optional => 1 },
                            distinct => { type => ARRAYREF | OBJECT,
                                          optional => 1 },
                            quote_identifiers => { type => BOOLEAN,
                                                   optional => 1 },
                          };

sub join
{
    my $self = shift;
    my %p = validate( @_, JOIN_SPEC );

    $p{join} ||= delete $p{tables};
    $p{join} = [ $p{join} ] unless Alzabo::Utils::is_arrayref( $p{join} );

    my @tables;

    if ( Alzabo::Utils::is_arrayref( $p{join}->[0] ) )
    {
        # flattens the nested structure and produces a unique set of
        # tables
        @tables = values %{ { map { $_ => $_ }
                              grep { Alzabo::Utils::safe_isa( $_, 'Alzabo::Table' ) }
                             map { @$_ } @{ $p{join} } } };
    }
    else
    {
        @tables = grep { Alzabo::Utils::safe_isa($_, 'Alzabo::Table') } @{ $p{join} };
    }

    if ( $p{distinct} )
    {
        $p{distinct} =
            Alzabo::Utils::is_arrayref( $p{distinct} ) ? $p{distinct} : [ $p{distinct} ];
    }

    if ( $p{order_by} )
    {
        $p{order_by} =
            Alzabo::Utils::is_arrayref( $p{order_by} )
            ? $p{order_by}
            : $p{order_by}
            ? [ $p{order_by} ]
            : undef;
    }

    # We go in this order:  $p{select}, $p{distinct}, @tables
    my @select_tables = ( $p{select} ?
                          ( Alzabo::Utils::is_arrayref( $p{select} ) ?
                            @{ $p{select} } : $p{select} ) :
                          $p{distinct} ?
                          @{ $p{distinct} } :
                          @tables );

    my $sql = Alzabo::Runtime::sqlmaker( $self, \%p );

    my @select_cols;
    if ( $p{distinct} )
    {
        my %distinct = map { $_ => 1 } @{ $p{distinct} };

        # hack so distinct is not treated as a function, just a
        # bareword in the SQL
        @select_cols = ( 'DISTINCT',
                         map { ( $_->primary_key,
                                 $_->prefetch ?
                                 $_->columns( $_->prefetch ) :
                                 () ) }
                         @{ $p{distinct} }
                       );

        foreach my $t (@select_tables)
        {
            next if $distinct{$t};
            push @select_cols, $t->primary_key;

            push @select_cols, $t->columns( $t->prefetch ) if $t->prefetch;
        }

        if ( $p{order_by} && $sql->distinct_requires_order_by_in_select )
        {
            my %select_cols = map { $_ => 1 } @select_cols;
            push @select_cols, grep { ref } @{ $p{order_by} };
        }

        @select_tables = ( @{ $p{distinct} }, grep { ! $distinct{$_} } @select_tables );
    }
    else
    {
        @select_cols =
            ( map { ( $_->primary_key,
                      $_->prefetch ?
                      $_->columns( $_->prefetch ) :
                      () ) }
              @select_tables );
    }

    $sql->select(@select_cols);

    $self->_join_all_tables( sql => $sql,
                             join => $p{join} );

    Alzabo::Runtime::process_where_clause( $sql, $p{where} ) if exists $p{where};

    Alzabo::Runtime::process_order_by_clause( $sql, $p{order_by} )
        if $p{order_by};

    $sql->limit( ref $p{limit} ? @{ $p{limit} } : $p{limit} ) if $p{limit};

    $sql->debug(\*STDERR) if Alzabo::Debug::SQL;
    print STDERR Devel::StackTrace->new if Alzabo::Debug::TRACE;

    my $statement = $self->driver->statement( sql => $sql->sql,
                                              bind => $sql->bind );

    if (@select_tables == 1)
    {
        return Alzabo::Runtime::RowCursor->new
                   ( statement => $statement,
                     table => $select_tables[0]->real_table,
                   );
    }
    else
    {
        return Alzabo::Runtime::JoinCursor->new
                   ( statement => $statement,
                     tables => [ map { $_->real_table } @select_tables ],
                   );
    }
}

sub row_count
{
    my $self = shift;
    my %p = @_;

    return $self->function( select => Alzabo::Runtime::sqlmaker( $self, \%p )->COUNT('*'),
                            %p,
                          );
}

sub function
{
    my $self = shift;
    my %p = @_;

    my $sql = $self->_select_sql(%p);

    my $method =
        Alzabo::Utils::is_arrayref( $p{select} ) && @{ $p{select} } > 1 ? 'rows' : 'column';

    $sql->debug(\*STDERR) if Alzabo::Debug::SQL;
    print STDERR Devel::StackTrace->new if Alzabo::Debug::TRACE;

    return $self->driver->$method( sql => $sql->sql,
                                   bind => $sql->bind );
}

sub select
{
    my $self = shift;

    my $sql = $self->_select_sql(@_);

    $sql->debug(\*STDERR) if Alzabo::Debug::SQL;
    print STDERR Devel::StackTrace->new if Alzabo::Debug::TRACE;

    return $self->driver->statement( sql => $sql->sql,
                                     bind => $sql->bind );
}

use constant _SELECT_SQL_SPEC => { join => { type => ARRAYREF | OBJECT,
                                             optional => 1 },
                                   tables => { type => ARRAYREF | OBJECT,
                                               optional => 1 },
                                   select => { type => SCALAR | ARRAYREF | OBJECT,
                                               optional => 1 },
                                   where => { type => ARRAYREF,
                                              optional => 1 },
                                   group_by => { type => ARRAYREF | HASHREF | OBJECT,
                                                 optional => 1 },
                                   order_by => { type => ARRAYREF | HASHREF | OBJECT,
                                                 optional => 1 },
                                   having => { type => ARRAYREF,
                                               optional => 1 },
                                   limit => { type => SCALAR | ARRAYREF,
                                              optional => 1 },
                                   quote_identifiers => { type => BOOLEAN,
                                                          optional => 1 },
                                 };

sub _select_sql
{
    my $self = shift;
    my %p = validate( @_, _SELECT_SQL_SPEC );

    $p{join} ||= delete $p{tables};
    $p{join} = [ $p{join} ] unless Alzabo::Utils::is_arrayref( $p{join} );

    my @tables;

    if ( Alzabo::Utils::is_arrayref( $p{join}->[0] ) )
    {
        # flattens the nested structure and produces a unique set of
        # tables
        @tables = values %{ { map { $_ => $_ }
                              grep { Alzabo::Utils::safe_isa( 'Alzabo::Table', $_ ) }
                              map { @$_ } @{ $p{join} } } };
    }
    else
    {
        @tables = grep { Alzabo::Utils::safe_isa( 'Alzabo::Table', $_ ) } @{ $p{join} };
    }

    my @funcs = Alzabo::Utils::is_arrayref( $p{select} ) ? @{ $p{select} } : $p{select};

    my $sql = ( Alzabo::Runtime::sqlmaker( $self, \%p )->
                select(@funcs) );

    $self->_join_all_tables( sql => $sql,
                             join => $p{join} );

    Alzabo::Runtime::process_where_clause( $sql, $p{where} )
        if exists $p{where};

    Alzabo::Runtime::process_group_by_clause( $sql, $p{group_by} )
        if exists $p{group_by};

    Alzabo::Runtime::process_having_clause( $sql, $p{having} )
        if exists $p{having};

    Alzabo::Runtime::process_order_by_clause( $sql, $p{order_by} )
        if exists $p{order_by};

    $sql->limit( ref $p{limit} ? @{ $p{limit} } : $p{limit} ) if $p{limit};

    return $sql;
}

use constant _JOIN_ALL_TABLES_SPEC => { join => { type => ARRAYREF },
                                        sql  => { isa => 'Alzabo::SQLMaker' } };

sub _join_all_tables
{
    my $self = shift;
    my %p = validate( @_, _JOIN_ALL_TABLES_SPEC );

    my @from;
    my @joins;

    # outer join given as only join
    $p{join} = [ $p{join} ] unless ref $p{join}->[0];

    # A structure like:
    #
    # [ [ $t_1 => $t_2 ],
    #   [ $t_1 => $t_3, $fk ],
    #   [ left_outer_join => $t_3 => $t_4 ],
    #   [ left_outer_join => $t_3 => $t_5, undef, [ $where_clause ] ]
    #
    if ( Alzabo::Utils::is_arrayref( $p{join}->[0] ) )
    {
        my %map;
        my %tables;

        foreach my $set ( @{ $p{join} } )
        {
            # we take some care not to change the contents of $set,
            # because the caller may reuse the variable being
            # referenced, and changes here could break that.

            # XXX - improve
            params_exception
                'The table map must contain only two tables per array reference'
                    if @$set > 5;

            my @tables;
            if ( ! ref $set->[0] )
            {
                $set->[0] =~ /^(right|left|full)_outer_join$/i
                    or params_exception "Invalid join type: $set->[0]";

                @tables = @$set[1,2];

                push @from, [ $1, @tables, @$set[3, 4] ];
            }
            else
            {
                @tables = @$set[0,1];

                push @from, grep { ! exists $tables{ $_->alias_name } } @tables;
                push @joins, [ @tables, $set->[2] ];
            }

            # Track the tables we've seen
            @tables{ $tables[0]->alias_name, $tables[1]->alias_name } = (1, 1);

            # Track their relationships
            push @{ $map{ $tables[0]->alias_name } }, $tables[1]->alias_name;
            push @{ $map{ $tables[1]->alias_name } }, $tables[0]->alias_name;
        }

        # just get one key to start with
        my ($key) = (each %tables)[0];
        delete $tables{$key};
        my @t = @{ delete $map{$key} };
        while (my $t = shift @t)
        {
            delete $tables{$t};
            push @t, @{ delete $map{$t} } if $map{$t};
        }

        logic_exception
            "The specified table parameter does not connect all the tables involved in the join"
                if keys %tables;
    }
    # A structure like:
    #
    # [ $t_1 => $t_2 => $t_3 => $t_4 ]
    #
    else
    {
        for (my $x = 0; $x < @{ $p{join} } - 1; $x++)
        {
            push @joins, [ $p{join}->[$x], $p{join}->[$x + 1] ];
        }

        @from = @{ $p{join} };
    }

    $p{sql}->from(@from);

    return unless @joins;

    foreach my $join (@joins)
    {
        $self->_join_two_tables( $p{sql}, @$join );
    }

    $p{sql}->subgroup_end;
}

sub _join_two_tables
{
    my $self = shift;
    my ($sql, $table_1, $table_2, $fk) = @_;

    my $op = $sql->last_op eq 'and' || $sql->last_op eq 'condition' ? 'and' : 'where';

    if ($fk)
    {
        unless ( $fk->table_from eq $table_1 && $fk->table_to eq $table_2 )
        {
            if ( $fk->table_from eq $table_2 && $fk->table_to eq $table_1 )
            {
                $fk = $fk->reverse;
            }
            else
            {
                params_exception
                    ( "The foreign key given to join together " .
                      $table_1->alias_name .
                      " and " . $table_2->alias_name .
                      " does not represent a relationship between those two tables" );
            }
        }
    }
    else
    {
        my @fk = $table_1->foreign_keys_by_table($table_2);

        logic_exception
            ( "The " . $table_1->name .
              " table has no foreign keys to the " .
              $table_2->name . " table" )
                unless @fk;

        logic_exception
            ( "The " . $table_1->name .
              " table has more than 1 foreign key to the " .
              $table_2->name . " table" )
                if @fk > 1;

        $fk = $fk[0];
    }

    foreach my $cp ( $fk->column_pair_names )
    {
        if ( $op eq 'where' )
        {
            # first time through loop only
            $sql->where;
            $sql->subgroup_start;
            $sql->condition( $table_1->column( $cp->[0] ), '=', $table_2->column( $cp->[1] ) );
        }
        else
        {
            $sql->$op( $table_1->column( $cp->[0] ), '=', $table_2->column( $cp->[1] ) );
        }
        $op = 'and';
    }
}

sub prefetch_all
{
    my $self = shift;

    $_->set_prefetch( $_->columns ) for $self->tables;
}

sub prefetch_all_but_blobs
{
    my $self = shift;

    $_->set_prefetch( grep { ! $_->is_blob } $_->columns ) for $self->tables;
}

sub prefetch_none
{
    my $self = shift;

    $_->set_prefetch() for $self->tables;
}

__END__