File Coverage

blib/lib/Alzabo/Runtime/Table.pm

Criterion	Covered	Total	%
statement	30	254	11.8
branch	0	108	0.0
condition	0	34	0.0
subroutine	10	36	27.7
pod	18	20	90.0
total	58	452	12.8

package Alzabo::Runtime::Table;

use strict;
use vars qw($VERSION);

use Alzabo::Exceptions ( abbr => [ qw( logic_exception not_nullable_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 Scalar::Util ();
use Tie::IxHash;

use base qw(Alzabo::Table);

$VERSION = 2.0;

sub insert
{
    my $self = shift;

    logic_exception "Can't make rows for tables without a primary key"
        unless $self->primary_key;

    my %p = @_;
    %p = validate( @_,
                   { ( map { $_ => { optional => 1 } } keys %p ),
                     values => { type => HASHREF, optional => 1 },
                     quote_identifiers => { type => BOOLEAN,
                                            optional => 1 },
                   },
                 );

    my $vals = delete $p{values} || {};

    my $schema = $self->schema;

    my @pk = $self->primary_key;
    foreach my $pk (@pk)
    {
        unless ( exists $vals->{ $pk->name } )
        {
            if ($pk->sequenced)
            {
                $vals->{ $pk->name } = $schema->driver->next_sequence_number($pk);
            }
            else
            {
                params_exception
                    ( "No value provided for primary key (" .
                      $pk->name . ") and no sequence is available." );
            }
        }
    }

    foreach my $c ($self->columns)
    {
        next if $c->is_primary_key;

        unless ( defined $vals->{ $c->name } || $c->nullable || defined $c->default )
        {
            not_nullable_exception
                ( error => $c->name . " column in " . $self->name . " table cannot be null.",
                  column_name => $c->name,
                  table_name  => $c->table->name,
                  schema_name => $c->table->schema->name,
                );
        }

        delete $vals->{ $c->name }
            if ! defined $vals->{ $c->name } && defined $c->default;
    }

    my @fk;
    @fk = $self->all_foreign_keys
        if $schema->referential_integrity;

    my $sql = ( Alzabo::Runtime::sqlmaker( $self->schema, \%p )->
                insert->
                into($self, $self->columns( sort keys %$vals ) )->
                values( map { $self->column($_) => $vals->{$_} } sort keys %$vals ) );

    my %id;

    $schema->begin_work if @fk;
    eval
    {
        foreach my $fk (@fk)
        {
            $fk->register_insert( map { $_->name => $vals->{ $_->name } } $fk->columns_from );
        }

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

        $self->schema->driver->do( sql => $sql->sql,
                                   bind => $sql->bind );

        foreach my $pk (@pk)
        {
            $id{ $pk->name } = ( defined $vals->{ $pk->name } ?
                                 $vals->{ $pk->name } :
                                 $schema->driver->get_last_id($self) );
        }

        # must come after call to ->get_last_id for MySQL because the
        # id will no longer be available after the transaction ends.
        $schema->commit if @fk;
    };
    if (my $e = $@)
    {
        eval { $schema->rollback };

        rethrow_exception $e;
    }

    return unless defined wantarray || $p{potential_row};

    return $self->row_by_pk( pk => \%id, %p );
}

sub insert_handle
{
    my $self = shift;

    logic_exception "Can't make rows for tables without a primary key"
        unless $self->primary_key;

    my %p = @_;
    %p = validate( @_,
                   { ( map { $_ => { optional => 1 } } keys %p ),
                     columns => { type => ARRAYREF, default => [] },
                     values  => { type => HASHREF, default => {} },
                     quote_identifiers => { type => BOOLEAN,
                                            optional => 1 },
                   },
                 );

    my %func_vals;
    my %static_vals;

    if ( $p{values} )
    {
        my $v = delete $p{values};
        while ( my ( $name, $val ) = each %$v )
        {
            if ( Alzabo::Utils::safe_isa( $val, 'Alzabo::SQLMaker::Function' ) )
            {
                $func_vals{$name} = $val;
            }
            else
            {
                $static_vals{$name} = $val
            }
        }
    }

    my $placeholder = $self->schema->sqlmaker->placeholder;

    my %cols;
    my %vals;
    # Get the unique set of columns and associated values
    foreach my $col ( @{ $p{columns} }, $self->primary_key )
    {
        $vals{ $col->name } = $placeholder;
        $cols{ $col->name } = 1;
    }

    foreach my $name ( keys %static_vals )
    {
        $vals{$name} = $placeholder;
        $cols{$name} = 1;
    }

    %vals = ( %vals, %func_vals );

    # At this point, %vals has each column's name and associated
    # value.  The value may be a placeholder or SQL function.

    $cols{$_} = 1 foreach keys %func_vals;

    foreach my $c ( $self->columns )
    {
        next if $c->is_primary_key  || $c->nullable || defined $c->default;

        unless ( $cols{ $c->name } )
        {
            not_nullable_exception
                ( error => $c->name . " column in " . $self->name . " table cannot be null.",
                  column_name => $c->name,
                  table_name  => $c->table->name,
                  schema_name => $c->table->schema->name,
                );
        }
    }

    my @columns = $self->columns( keys %vals );

    my $sql = ( Alzabo::Runtime::sqlmaker( $self->schema, \%p )->
                insert->
                into( $self, @columns )->
                values( map { $_ => $vals{ $_->name } } @columns ),
              );

    return Alzabo::Runtime::InsertHandle->new( table => $self,
                                               sql   => $sql,
                                               values  => \%static_vals,
                                               columns => \@columns,
                                               %p,
                                             );
}

sub row_by_pk
{
    my $self = shift;

    logic_exception "Can't make rows for tables without a primary key"
        unless $self->primary_key;

    my %p = @_;

    my $pk_val = $p{pk};

    my @pk = $self->primary_key;

    params_exception
        'Incorrect number of pk values provided.  ' . scalar @pk . ' are needed.'
            if ref $pk_val && @pk != scalar keys %$pk_val;

    if (@pk > 1)
    {
        params_exception
            ( 'Primary key for ' . $self->name . ' is more than one column.' .
              '  Please provide multiple key values as a hashref.' )
                unless ref $pk_val;

        foreach my $pk (@pk)
        {
            params_exception 'No value provided for primary key ' . $pk->name . '.'
                unless defined $pk_val->{ $pk->name };
        }
    }

    return $self->_make_row( %p,
                             table => $self,
                           );
}

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

    my $class = $p{row_class} ? delete $p{row_class} : $self->_row_class;

    return $class->new(%p);
}

sub _row_class { 'Alzabo::Runtime::Row' }

sub row_by_id
{
    my $self = shift;
    my %p = @_;
    validate( @_, { row_id => { type => SCALAR },
                    ( map { $_ => { optional => 1 } } keys %p ) } );

    my (undef, undef, %pk) = split ';:;_;:;', delete $p{row_id};

    return $self->row_by_pk( %p, pk => \%pk );
}

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

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

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

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

    return $self->_cursor_by_sql( %p, sql => $sql );
}

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

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

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

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

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

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

    my @return = $self->schema->driver->one_row( sql => $sql->sql,
                                                 bind => $sql->bind )
        or return;

    my @pk = $self->primary_key;

    my (%pk, %prefetch);

    @pk{ map { $_->name } @pk } = splice @return, 0, scalar @pk;

    # Must be some prefetch pieces
    if (@return)
    {
        @prefetch{ $self->prefetch } = @return;
    }

    return $self->row_by_pk( pk => \%pk,
                             prefetch => \%prefetch,
                           );
}

sub all_rows
{
    my $self = shift;

    my $sql = $self->_make_sql;

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

    return $self->_cursor_by_sql( @_, sql => $sql );
}

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

    logic_exception "Can't make rows for tables without a primary key"
        unless $self->primary_key;

    my $sql = ( Alzabo::Runtime::sqlmaker( $self->schema, \%p )->
                select( $self->primary_key,
                        $self->prefetch ? $self->columns( $self->prefetch ) : () )->
                from( $self ) );

    return $sql;
}

sub _cursor_by_sql
{
    my $self = shift;

    my %p = @_;
    validate( @_, { sql => { isa => 'Alzabo::SQLMaker' },
                    order_by => { type => ARRAYREF | HASHREF | OBJECT,
                                  optional => 1 },
                    limit => { type => SCALAR | ARRAYREF,
                               optional => 1 },
                    ( map { $_ => { optional => 1 } } keys %p ) } );

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

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

    my $statement = $self->schema->driver->statement( sql => $p{sql}->sql,
                                                      bind => $p{sql}->bind,
                                                      limit => $p{sql}->get_limit );

    return Alzabo::Runtime::RowCursor->new( statement => $statement,
                                            table => $self,
                                          );
}

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

    logic_exception "Can't make rows for tables without a primary key"
        unless $self->primary_key;

    my $class = $p{row_class} ? delete $p{row_class} : $self->_row_class;

    return $class->new( %p,
                        state => 'Alzabo::Runtime::RowState::Potential',
                        table => $self,
                      );
}

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

    my $count = Alzabo::Runtime::sqlmaker( $self->schema, \%p )->COUNT('*');

    return $self->function( select => $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->schema->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->schema->driver->statement( sql => $sql->sql,
                                             bind => $sql->bind );
}

use constant
    _SELECT_SQL_SPEC => { select => { type => SCALAR | ARRAYREF | OBJECT },
                          where  => { type => ARRAYREF | OBJECT,
                                      optional => 1 },
                          order_by => { type => ARRAYREF | HASHREF | OBJECT,
                                        optional => 1 },
                          group_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 );

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

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

    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;
}

sub set_prefetch
{
    my $self = shift;

    $self->{prefetch} = $self->_canonize_prefetch(@_);
}

sub _canonize_prefetch
{
    my $self = shift;

    validate_pos( @_, ( { isa => 'Alzabo::Column' } ) x @_ );

    foreach my $c (@_)
    {
        params_exception "Column " . $c->name . " doesn't exist in $self->{name}"
            unless $self->has_column( $c->name );
    }

    return [ map { $_->name } grep { ! $_->is_primary_key } @_ ];
}

sub prefetch
{
    my $self = shift;

    return ref $self->{prefetch} ? @{ $self->{prefetch} } : ();
}

sub add_group
{
    my $self = shift;

    validate_pos( @_, ( { isa => 'Alzabo::Column' } ) x @_ );

    my @names = map { $_->name } @_;
    foreach my $col (@_)
    {
        params_exception "Column " . $col->name . " doesn't exist in $self->{name}"
            unless $self->has_column( $col->name );

        next if $col->is_primary_key;
        $self->{groups}{ $col->name } = \@names;
    }
}

sub group_by_column
{
    my $self = shift;
    my $col = shift;

    return exists $self->{groups}{$col} ? @{ $self->{groups}{$col} } : $col;
}

my $alias_num = '000000000';
sub alias
{
    my $self = shift;

    my $clone;
    %$clone = %$self;

    bless $clone, ref $self;

    $clone->{alias_name} = $self->name . ++$alias_num;
    $clone->{real_table} = $self;

    $clone->{columns} = Tie::IxHash->new( map { $_->name => $_ } $self->columns );

    # Force clone of primary key columns right away.
    $clone->column($_) foreach map { $_->name } $self->primary_key;

    return $clone;
}

#
# Since its unlikely that a user will end up needing clones of more
# than 1-2 columns each time an alias is used, we only make copies as
# needed.
#
sub column
{
    my $self = shift;

    # I'm an alias, make an alias column
    if ( $self->{alias_name} )
    {
        my $name = shift;
        my $col = $self->SUPER::column($name);

        # not previously cloned
        unless ( $col->table eq $self )
        {
            # replace our copy of this column with a clone
            $col = $col->alias_clone( table => $self );
            my $index = $self->{columns}->Indices($name);
            $self->{columns}->Replace( $index, $col, $name );

            Scalar::Util::weaken( $col->{table} );

            delete $self->{pk_array} if $col->is_primary_key;
        }

        return $col;
    }
    else
    {
        return $self->SUPER::column(@_);
    }
}

sub alias_name
{
    # intentionally don't call $_[0]->name for a noticeable
    # performance boost
    return $_[0]->{alias_name} || $_[0]->{name};
}

sub real_table
{
    return $_[0]->{real_table} || $_[0];
}

# This gets called a _lot_ so doing this sort of 'memoization' helps
sub primary_key
{
    my $self = shift;

    $self->{pk_array} ||= [ $self->SUPER::primary_key ];

    return ( wantarray ?
             @{ $self->{pk_array} } :
             $self->{pk_array}->[0]
           );
}

1;

__END__