File Coverage

blib/lib/DataCube.pm

Criterion	Covered	Total	%
statement	37	39	94.8
branch			n/a
condition			n/a
subroutine	13	13	100.0
pod			n/a
total	50	52	96.1

package DataCube;

use 5.008008;
our $VERSION = '0.01';

use strict;
use warnings;

use Fcntl;
use URI::file;
use Digest::MD5;
use Time::HiRes;
use Data::Dumper;
use Cwd qw(getcwd);
use Storable qw(nstore);
use Scalar::Util qw(reftype);

use constant {
    read_only      => O_RDONLY,
    write_only     => O_WRONLY,
    create_write   => O_CREAT|O_WRONLY,
    vip_read_write => O_CREAT|O_WRONLY|O_EXCL,
};

use DataCube::Cube;
use DataCube::Schema;
use DataCube::FileUtils;
use DataCube::CubeStore;
use DataCube::Controller;
use DataCube::PathWalker;
use DataCube::Connection;
use DataCube::FileUtils::FileReader;

sub new {
    my($class,@opts,%opts) = @_;
    datacube_opts_collection: {
        %opts         = @opts    and last datacube_opts_collection if @_  > 2 && @_ % 2;
        $opts{schema} = $opts[0] and last datacube_opts_collection if @_ == 2;
    }
    my $self = bless {}, ref($class) || $class;
    return $self->retrieve($opts[0]) if defined($opts[0]) && -f($opts[0]) && @opts == 1;
    return $self unless @opts;
    $self->{cube_store} = DataCube::CubeStore->new;
    schema_initialization: {
        last schema_initialization unless
                       $opts{schema}
                && ref($opts{schema})
                && ref($opts{schema}) =~ /^datacube::schema$/i;
        die "DataCube(new : schema init):\nyour schema does not contain any measures\n$!\n"
            unless $opts{schema}->{measures};
        $opts{schema}->initialize;
        $self->build_lattice_tables (
            schema => $opts{schema},
        );
    }
    $self->{updater}   = DataCube::MeasureUpdater->new($opts{schema});
    my $base_cube_name = $self->{controller}->{cube_stats}->{base_cube_name};
    $self->{meta_data}->{system}->{base_cube_name} = $base_cube_name;
    $self->{meta_data}->{system}->{base_schema} = $opts{schema};
    $self->{base_cube} = $self->{cube_store}->{cubes}->{$base_cube_name};
    return $self;
}

# change to modify:
# ------------------------------------------------------------------------------
# DataCube::insert
# DataCube::reset_measures
# DataCube::MeasureUpdater::update
# DataCube::Cube::describe
# DataCube::Cube::to_table
# Warehouse::Builder::*
# ------------------------------------------------------------------------------

sub insert {
    my($self,$data) = @_;
    my @keys;
    push @keys, $data->{$_} for @{$self->{base_cube}->{schema}->{fields}};
    my $key = join("\t",@keys);
    measure_update:
    for(@{$self->{base_cube}->{schema}->{measures}}){

        if($_->[0] eq 'key_count'){
            ++$self->{base_cube}->{cube}->{$key}->{key_count};
            next measure_update;
        }

        my $data_key = $data->{$_->[1]};

        if($_->[0] eq 'count'){
            $self->{base_cube}->{cube}->{$key}->{count}->{$_->[1]}->{$data_key} = undef;
            next measure_update;
        }

        if($_->[0] eq 'multi_count'){
            ++$self->{base_cube}->{cube}->{$key}->{multi_count}->{$_->[1]}->{$data_key};
            next measure_update;
        }

        if($_->[0] eq 'sum'){
            $self->{base_cube}->{cube}->{$key}->{sum}->{$_->[1]} += $data_key;
            next measure_update;
        }

        if($_->[0] eq 'average'){
            $self->{base_cube}->{cube}->{$key}->{average}->{$_->[1]}->{sum_total}   += $data_key;
            ++$self->{base_cube}->{cube}->{$key}->{average}->{$_->[1]}->{observations};
            next measure_update;
        }

        if($_->[0] eq 'max'){
            $self->{base_cube}->{cube}->{$key}->{max}->{$_->[1]} = $data_key unless defined $self->{base_cube}->{cube}->{$key}->{max}->{$_->[1]};
            $self->{base_cube}->{cube}->{$key}->{max}->{$_->[1]} = $data_key if $data_key > $self->{base_cube}->{cube}->{$key}->{max}->{$_->[1]};
            next measure_update;
        }

        if($_->[0] eq 'min'){
            $self->{base_cube}->{cube}->{$key}->{min}->{$_->[1]} = $data_key unless defined $self->{base_cube}->{cube}->{$key}->{min}->{$_->[1]};
            $self->{base_cube}->{cube}->{$key}->{min}->{$_->[1]} = $data_key if $data_key < $self->{base_cube}->{cube}->{$key}->{min}->{$_->[1]};
            next measure_update;
        }

        if($_->[0] eq 'product'){
            $self->{base_cube}->{cube}->{$key}->{product}->{$_->[1]}  = 1 unless defined $self->{base_cube}->{cube}->{$key}->{product}->{$_->[1]};
            $self->{base_cube}->{cube}->{$key}->{product}->{$_->[1]} *= $data_key;
            next measure_update;
        }

    }
    return $key;
}

sub store {
   my($self,$path) = @_;
   Storable::nstore($self,$path);
   return $self;
}

sub retrieve {
    my($self,$path) = @_;
    return $_[0] = Storable::retrieve($path);
}

sub clone {
    my($self,$clone) = @_;
    return Storable::thaw(Storable::freeze($self));
}

sub get_measures {
    my($self,$data) = @_;
    my @keys;
    my @values;
    for(@{$self->{base_cube}->{schema}->{fields}}){
        if(defined $data->{$_}){
            push @keys,   $_;
            push @values, $data->{$_};
        }
    }
    return unless @keys;
    my $table   = join("\t",@keys);
    my $datakey = join("\t",@values);
    return $self->cube_store->fetch($table)->data->{$datakey};
}

sub get_measures_by_id {
    my($self,$key) = @_;
    return $self->{base_cube}->{cube}->{$key};
}

sub query_measures {
    my($self,@table) = @_;

    push @table, sub{1}
        unless @table
        && ref($table[$#table])
        && ref($table[$#table]) =~ /^code$/i;

    my $func = pop(@table);
    my $name = join("\t", sort @table);

    my $table = $self->cube_store->fetch($name);

    return unless $table;

    my $data  = $table->{cube};

    my @results;

    my @fields = $table->schema->fields;

    for(keys %$data){
        my $key = $_;
        my @key = split/\t/,$key,-1;
        my %key = map { $fields[$_] => $key[$_] } (0 .. $#fields);
        my $record =  { datakey => \%key, measures => $data->{$_} };
        push @results, $record if $func->($record);
    }

    return @results;
}


sub delete {
    my($self,$data) = @_;

    die "DataCube(delete):\n" .
        "cannot delete rows in a cube that has been rolled up\n$!\n"
        if $self->has_been_rolled_up;

    my @values;
    for(@{$self->{base_cube}->{schema}->{fields}}){

        die "DataCube(delete):\n" .
            "please pass a hash reference with defined entries in the base table:\n"
        unless defined $data->{$_};

        push @values, $data->{$_};
    }

    my $datakey = join("\t",@values);
    delete $self->{base_cube}->{cube}->{$datakey};
    return $self;
}

sub reset {
    my($self) = @_;
    for(keys %{$self->cube_store->cubes}){
        my $cube = $self->cube_store->fetch($_);
        $cube->reset;
    }
    $self->{meta_data}->{system}->{has_been_rolled_up} = 0;
    return $self;
}

sub unroll {
    my($self) = @_;
    for(keys %{$self->cube_store->cubes}){
        my $cube = $self->cube_store->fetch($_);
        $cube->reset unless $cube->schema->name eq $self->base_cube_name;
    }
    delete $self->{meta_data}->{system}->{has_been_rolled_up};
    return $self;
}


sub reset_measures {
    my($self,$data) = @_;

    die "DataCube(reset_measures):\n" .
        "cannot reset measures in a cube that has been rolled up\n$!\n"
        if $self->has_been_rolled_up;

    my @values;
    for(@{$self->{base_cube}->{schema}->{fields}}){

        die "DataCube(reset_measures):\n" .
            "please pass a hash reference with defined entries in the base table:\n"
        unless defined $data->{$_};

        push @values, $data->{$_};
    }
    my $table = $self->base_table;
    my $key   = join("\t",@values);

    measure_reset:
    for(@{$self->{base_cube}->{schema}->{measures}}){
        if($_->[0] eq 'key_count'){
            $self->{base_cube}->{cube}->{$key}->{key_count} = 0;
            next measure_reset;
        }
        if($_->[0] eq 'count'){
            $self->{base_cube}->{cube}->{$key}->{count}->{$_->[1]} = {};
            next measure_reset;
        }
        if($_->[0] eq 'multi_count'){
            $self->{base_cube}->{cube}->{$key}->{multi_count}->{$_->[1]} = {};
            next measure_reset;
        }
        if($_->[0] eq 'sum'){
            $self->{base_cube}->{cube}->{$key}->{sum}->{$_->[1]} = 0;
            next measure_reset;
        }
        if($_->[0] eq 'product'){
            $self->{base_cube}->{cube}->{$key}->{product}->{$_->[1]}  = 1;
            next measure_reset;
        }
        if($_->[0] eq 'max'){
            $self->{base_cube}->{cube}->{$key}->{max}->{$_->[1]} = undef;
            next measure_reset;
        }
        if($_->[0] eq 'min'){
            $self->{base_cube}->{cube}->{$key}->{min}->{$_->[1]} = undef;
            next measure_reset;
        }
        if($_->[0] eq 'average'){
            $self->{base_cube}->{cube}->{$key}->{average}->{$_->[1]}->{sum_total} = 0;
            $self->{base_cube}->{cube}->{$key}->{average}->{$_->[1]}->{observations} = 0;
            next measure_reset;
        }

    }
    return $self;
}


sub decrement_key_count {
    my($self,$data) = @_;

    die "DataCube(decrement_key_count):\n" .
        "cannot decrement_key_count in a cube that has been rolled up\n$!\n"
        if $self->has_been_rolled_up;

    my $key = join("\t", map { $data->{$_} }  @{$self->{base_cube}->{schema}->{fields}} );
    --$self->{base_cube}->{cube}->{$key}->{key_count};
    return $self;
}

sub decrement_multi_count {
    my($self,$field,$data) = @_;

    die "DataCube(decrement_multi_count):\n" .
        "cannot decrement_multi_count in a cube that has been rolled up\n$!\n"
        if $self->has_been_rolled_up;

    my $key = join("\t", map { $data->{$_} }  @{$self->{base_cube}->{schema}->{fields}} );
    --$self->{base_cube}->{cube}->{$key}->{multi_count}->{$field}->{$data->{$field}};
    return $self;
}

sub drop_count {
    my($self,$field,$data) = @_;

    die "DataCube(drop_count):\n" .
        "cannot drop_count in a cube that has been rolled up\n$!\n"
        if $self->has_been_rolled_up;

    my $key = join("\t", map { $data->{$_} }  @{$self->{base_cube}->{schema}->{fields}} );
    delete $self->{base_cube}->{cube}->{$key}->{count}->{$field}->{$data->{$field}};
    return $self;
}

sub add_meta_data {
    my($self,%meta_data) = @_;
    $self->{meta_data}->{user_generated}->{$_} = $meta_data{$_} for keys %meta_data;
    return $self;
}


sub load_data_infile {
    my($self,$file,$line) = @_;
    my $reader = DataCube::FileUtils::FileReader->new->read($file);
    $self->insert($line) while $line = $reader->nextrow_hashref;
    return $self;
}

sub report_html {
    my($self,$dir,$target) = @_;
    return $self->report_commited_html($dir,$target) if  $dir && $target && -d($dir) && -d($target);
    mkdir($dir);
    for(keys %{$self->cube_store->cubes}){
        my $cube = $self->cube_store->fetch($_);
        $cube->report_html($dir);
    }
    return $self;
}

sub report {
    my($self,$dir,$target) = @_;
    return $self->report_commited($dir,$target) if  $dir && $target && -d($dir) && -d($target);
    mkdir($dir);
    for(keys %{$self->cube_store->cubes}){
        my $cube = $self->cube_store->fetch($_);
        $cube->report($dir);
    }
    return $self;
}

sub report_commited {
    my($self,$source,$target) = @_;
    mkdir($target);
    my $connection = DataCube::Connection->new($source);
    $connection->report($target);
    return $self;
}

sub report_html_commited {
    my($self,$source,$target) = @_;
    mkdir($target);
    my $connection = DataCube::Connection->new($source);
    $connection->report_html($target);
    return $self;
}

sub sync {
    my($self,$target) = @_;
    my $connection = DataCube::Connection->new($target);
    $connection->sync;
    return $self;
}

sub lazy_rollup {
    my($self,%opts) = @_;

    die "DataCube(lazy_rollup):\ncant lazy_rollup on a cube that has already been rolled up\n$!\n"
        if $self->has_been_rolled_up;

    $self->initialize_lazy_rollup unless $self->can_lazy_rollup;
    if($self->{lazy_rollup_list} && ! @{$self->{lazy_rollup_list}}){
        delete $self->{lazy_rollup_list};
        return;
    }
    if($self->{lazy_rollup_list}->[0] eq $self->base_cube_name) {
        shift @{$self->{lazy_rollup_list}};
        return $self->base_cube;
    }
    my $source = [split/\t/,$self->{meta_data}->{system}->{base_cube_name},-1];
    my $target = [split/\t/,shift(@{$self->{lazy_rollup_list}}),-1];
    return $self->rollup_from(
        parent => $source,
        child  => $target,
    );
}

sub initialize_lazy_rollup {
    my($self) = @_;
    my @pending = sort {$a <=> $b} keys %{$self->{controller}->{cube_stats}->{field_count}};
    for(@pending){
        push @{$self->{lazy_rollup_list}},
            sort @{$self->{controller}->{cube_stats}->{field_count}->{$_}};
    }
    return $self;
}

sub can_lazy_rollup {
    my($self) = @_;
    return $self->{lazy_rollup_list};
}

sub how_many_active_cubes {
    my($self) = @_;
    my $total = 0;
    for(keys %{$self->{cube_store}->cubes}){
        $total++ if scalar(keys(%{$self->{cube_store}->{cubes}->{$_}->{cube}}));
    }
    return $total;
}

sub rollup_from {
    my($self,%opts) = @_;

    $opts{child}  = [split/\t/,$opts{child} ,-1] unless ref($opts{child});
    $opts{parent} = [split/\t/,$opts{parent},-1] unless ref($opts{parent});

    my @parent = sort @{$opts{parent}};
    my @child  = sort @{$opts{child}};

    my %child = map { $_ => undef } @child;

    my $child_cube  = $self->{cube_store}->{cubes}->{join("\t",@child)};
    my $parent_cube = $self->{cube_store}->{cubes}->{join("\t",@parent)};

    die "DataCube(rollup_from):\ncould not locate child cube:@child\n"
        unless $child_cube && $child_cube->{schema};

    my @index_map = ();
    for(my $i = 0; $i < @parent; $i++){
        push @index_map, $i if exists $child{$parent[$i]};
    }

    die "DataCube(rollup_from):\nfound no compatible index mapping\n@parent\n@child\n"
        unless(@index_map || ($#child == 0 && $child[0] eq 'overall'));

    my $cube    = $child_cube->new(schema => $child_cube->{schema});
    my $updater = DataCube::MeasureUpdater->new( $cube->{schema} );

    my $target_data = $cube->{cube};
    my $source_data = $parent_cube->{cube};


    for(keys %$source_data){

        my $old_key = $_;
        my @old_key = split/\t/,$old_key,-1;

        local $SIG{__WARN__} = sub {
                die "DataCube(rollup_from | warnings):\n".
                    "caught a fatal exception here:\n$_[0]\n" .
                    '-' x 80 . "\n" .
                    join("\n",
                        "old_key:      $old_key",
                        'index_map:    ' . join(", ",@index_map) ,
                        "parent_cube:  $parent_cube",
                        "child_cube:   $child_cube"),"\n";
        };

        my $new_key = join("\t",@old_key[@index_map]);


        $updater->update (
            target     => $target_data,
            source     => $source_data,
            source_key => $old_key,
            target_key => $new_key,
        );
    }
    return $cube;
}

sub rollup {
    my($self,%opts)    = @_;

    die "DataCube(rollup):\ncant rollup twice!\n"
        if $self->has_been_rolled_up;

    my @measures       = @{$self->{meta_data}->{system}->{base_schema}->{measures}};
    my @pending_levels = sort { $b <=> $a } keys %{$self->{controller}->{cube_stats}->{field_count}};

    splice(@pending_levels, 0 ,1);

    for(@pending_levels){
        my $level = $_;
        my @next_cubes = @{$self->{controller}->{cube_stats}->{field_count}->{$level}};

        for(@next_cubes){
            my $next_cube_name = $_;

            my $possible_parents = $self->{controller}->{cube_stats}->{possible_parents}->{$next_cube_name};

            unless($possible_parents && ref($possible_parents) =~ /^array$/i){
                die "DataCube(rollup : possible_parents):\n" .
                    "cant find possible parent list for:\nnext_cube_name:\n$next_cube_name" .
                    "\nat level:\n$level\n";
            }

            my @possible_parents = @$possible_parents;

            my $best_parent_name = $possible_parents[0];
            my $best_size        = scalar(keys %{$self->{cube_store}->{cubes}->{$possible_parents[0]}->{cube}});

            for(my $i = 1; $i < @possible_parents; $i++){
                my $new_size = scalar(keys %{$self->{cube_store}->{cubes}->{$possible_parents[$i]}->{cube}});
                ($best_parent_name, $best_size) = ($possible_parents[$i], $new_size) if $new_size < $best_size;
            }

            my $next_cube   = $self->{cube_store}->fetch($next_cube_name);
            my $best_parent = $self->{cube_store}->fetch($best_parent_name);

            my @best_parent_index_map = ();

            if($next_cube_name ne 'overall'){
                my @parent_fields = split/\t/,$best_parent_name;
                my $child_fields = $next_cube->{schema}->{field_names};
                index_map_collection:
                for(my $i = 0; $i < @parent_fields; $i++){
                    next index_map_collection unless exists $child_fields->{$parent_fields[$i]};
                    push @best_parent_index_map, $i;
                }
            }

            my $updater = DataCube::MeasureUpdater->new( $next_cube->{schema} );

            my $target_data = $next_cube->{cube};
            my $source_data = $best_parent->{cube};

            for(keys %$source_data){

                my $old_key = $_;
                my @old_key = split/\t/,$old_key,-1;
                my $new_key = join("\t",@old_key[@best_parent_index_map]);

                $updater->update (
                    target     => $target_data,
                    source     => $source_data,
                    source_key => $old_key,
                    target_key => $new_key,
                );
            }
        }
    }
    $self->{meta_data}->{system}->{has_been_rolled_up} = 1;
    return $self;
}

sub commit {
    my($self,$target) = @_;
    mkdir($target);
    my $cubes = $self->{cube_store}->cubes;
    $cubes->{$_}->commit($target) for keys %$cubes;
    return $self;
}

sub lazy_rc {
    my($self,$target) = @_;
	while(my $next_cube = $self->lazy_rollup){
        $next_cube->commit($target);
	}
    return $self;
}

sub build_lattice_tables {
    my($self,%opts) = @_;
    my $schema      = $opts{schema};
    $schema->check_conflicts;
    my $confine = $schema->is_confined;
    my @hierarchies = @{$schema->{hierarchies}};
    my @lattice;
    for(my $i = 0; $i < @hierarchies; $i++){
        my @hierarchy = @{$hierarchies[$i]};
        for(my $j = 0; $j < @hierarchy; $j++){
            $lattice[$i][$j] = [@hierarchy[0..$j]];
        }
    }

    my $path_walker = DataCube::Pathwalker->new(
        lattice => [@lattice],
    );

    path_walk:
    while(my $next_path = $path_walker->next_path){

        my $next_schema = DataCube::Schema->new;

        $next_schema->add_measure(@$_)          for @{$schema->{measures}};
        $next_schema->add_computed_measure(@$_) for @{$schema->{computed_measures}};

        for(@$next_path){
            my @path = grep{defined}@$_;
            $next_schema->add_strict_hierarchy(@path) if @path;
        }

        my $cube = DataCube::Cube->new(
            schema  => $next_schema,
        );

        $cube->{is_the_base_table} = 1 if $cube->{schema}->{name} eq $schema->{name};

        if($schema->{confine_to_base}){
            $self->{cube_store}->add_cube($cube) and last path_walk if $cube->{is_the_base_table};
            next path_walk;
        }

        if($schema->{lattice_point_names} && defined($schema->{lattice_point_names}->{$cube->{schema}->{name}}) ){
            $cube->{schema}->{lattice_point_name} =
                $schema->{lattice_point_names}->{$cube->{schema}->{name}};
        }

        if(defined $confine){
            next path_walk if $confine ne $cube->{schema}->{name};
            $self->{cube_store}->add_cube($cube) and last path_walk
                if $confine eq $cube->{schema}->{name};
        }

        if( $schema->{lattice_point_filters} ){
            my @fields  = split/\t/,$cube->{schema}->{name};
            my @filters = @{ $schema->{lattice_point_filters} };
            for( @filters ){
                next unless ref( $_ ) && ref( $_ ) eq 'CODE';
                next path_walk if $_->( @fields ) && ! $cube->{is_the_base_table};
            }
        }

        if( $schema->{asserted_lattice_points} && ref($schema->{asserted_lattice_points}) ){
            $self->{cube_store}->add_cube($cube)
                if exists
                    $schema->{asserted_lattice_points}->{$cube->{schema}->{name}}   ||
                    $cube->{schema}->{name} eq $schema->{name};
        } else {
            $self->{cube_store}->add_cube($cube)
                unless exists
                    $schema->{suppressed_lattice_points}->{$cube->{schema}->{name}} &&
                    $cube->{schema}->{name} ne $schema->{name};
        }

    }
    $self->{controller} = DataCube::Controller->new_from_datacube($self);
    return $self;

}

sub merge_with {
    my($self,@data_cubes) = @_;
    my $base_cube_name = $self->{meta_data}->{system}->{base_cube_name};
    my $base_schema    = $self->{cube_store}->fetch($base_cube_name)->{schema};
    my $self_cubes     = $self->{cube_store}->cubes;

    for(@data_cubes){
        my $data_cube    = $_;
        my $parity_check = $data_cube->has_been_rolled_up + $self->has_been_rolled_up;

        die "DataCube(merge_with):\ncant merge a cube which has been rolled up with one that has not\n" if $parity_check % 2;
        my $new_cubes = $data_cube->{cube_store}->cubes;

        for(keys %$self_cubes){ die "DataCube(merge_with : self_cubes):\ncube name mismatch:\n$_\n$!\n" unless defined $new_cubes->{$_} }
        for(keys %$new_cubes) { die "DataCube(merge_with : new_cubes):\ncube name mismatch:\n$_\n$!\n"  unless defined $self_cubes->{$_}}

        for(keys %$self_cubes){
            $self_cubes->{$_}->merge_with( $new_cubes->{$_} );
        }
    }

    return $self;
}

sub isa_copy_of {
    my($self,$cube) = @_;
    my @queue = ($self,$cube);
    equality_check:
    while(1){
        last equality_check unless @queue;
        my($source,$target) = splice(@queue,0,2);
        return if (ref($source) && !ref($target)) || (!ref($source) && ref($target));
        next equality_check unless defined($source) && defined($target);
        if(!ref($source) && !ref($target)){
            return unless $source eq $target;
            next equality_check;
        }
        if(reftype($source) =~ /^array$/i && reftype($target) =~ /^array$/i){
            return unless @$source == @$target;
            push @queue, ($source->[$_],$target->[$_]) for (0..$#$source);
            next equality_check;
        }
        if(reftype($source) =~ /^hash$/i && reftype($target) =~ /^hash$/i){
            return unless keys %$source == keys %$target;
            push @queue, ($source->{$_},$target->{$_}) for keys %$source;
            next equality_check;
        }
    }
    return 1;
}

sub has_been_rolled_up {
    my($self) = @_;
    return $self->{meta_data}->{system}->{has_been_rolled_up} || 0;
}

sub rollup_by_level {
    my($self,%opts) = @_;
    $self->initialize_rollup_by_level unless $self->can_rollup_by_level;
    unless($self->{level_rollup_state}->{pending_levels}){
        delete $self->{level_rollup_state};
        return;
    }
    my $cube_name = $self->{level_rollup_state}->{current_cube};
    my $parents   = $self->{controller}->{cube_stats}->{possible_parents}->{$cube_name};
    unless($parents) {
        $self->nominate_next_level_cube;
        return $self->{base_cube};
    }
    if(my $last_name = $self->{level_rollup_state}->{current_cleanup}){
        unless($self->{level_rollup_state}->{parent_ref_count}->{$last_name}){
            delete $self->{cube_store}->{cubes}->{$last_name}->{cube};
            $self->{cube_store}->{cubes}->{$last_name}->{cube} = {};
        }
    }
    my $parent = $self->{level_rollup_state}->{possible_parents}->{$cube_name};
    $self->{cube_store}->{cubes}->{$cube_name} =
    $self->rollup_from(
        parent => $parent,
        child  => $cube_name,
    );
    $self->{level_rollup_state}->{parent_ref_count}->{$parent}--;
    my $cleanup =
                 0 >= $self->{level_rollup_state}->{parent_ref_count}->{$parent}
        && $parent ne $self->{meta_data}->{system}->{base_cube_name};
    if($cleanup){
        delete $self->{level_rollup_state}->{parent_ref_count}->{$parent};
        delete $self->{cube_store}->{cubes}->{$parent}->{cube};
        $self->{cube_store}->{cubes}->{$parent}->{cube} = {};
    }
    $self->nominate_next_level_cube;
    $self->{level_rollup_state}->{current_cleanup} = $cube_name;
    return $self->{cube_store}->fetch($cube_name);
}

sub nominate_next_level_cube {
    my($self) = @_;
    my $current_level = $self->{level_rollup_state}->{current_level};
    my @current_level = @{$self->{level_rollup_state}->{pending_levels}->{$current_level}};

    unless(@current_level){
        delete $self->{level_rollup_state}->{pending_levels}->{$current_level};
        my @pending_levels = sort {$b <=> $a} keys %{$self->{level_rollup_state}->{pending_levels}};
        unless(@pending_levels){
            delete $self->{level_rollup_state}->{pending_levels};
            return $self;
        }
        my $child = shift(@{$self->{level_rollup_state}->{pending_levels}->{$pending_levels[0]}});
        $self->{level_rollup_state}->{current_cube}   = $child;
        $self->{level_rollup_state}->{current_parent} = $self->{level_rollup_state}->{possible_parents}->{$child};
        $self->{level_rollup_state}->{current_level}  = $pending_levels[0];

    } else {
        my $child = shift(@{$self->{level_rollup_state}->{pending_levels}->{$current_level}});
        $self->{level_rollup_state}->{current_cube}   = $child;
        $self->{level_rollup_state}->{current_parent} = $self->{level_rollup_state}->{possible_parents}->{$child};
    }
    return $self;
}

sub initialize_rollup_by_level {
    my($self) = @_;
    my $possible = $self->{controller}->{cube_stats}->{possible_parents};
    for(keys %$possible){
        my($child,@parents) = ($_,@{$possible->{$_}});
        $self->{level_rollup_state}->{possible_parents}->{$child} = $parents[0];
        $self->{level_rollup_state}->{parent_ref_count}->{$parents[0]}++;
    }
    for(keys %{$self->{controller}->{cube_stats}->{field_count}}){
        my $level = $_;
        my @cubes = @{$self->{controller}->{cube_stats}->{field_count}->{$level}};
        $self->{level_rollup_state}->{pending_levels}->{$level} = [@cubes];
    }
    my @pending_levels = sort {$b <=> $a} keys %{$self->{level_rollup_state}->{pending_levels}};
    $self->{level_rollup_state}->{current_cube}  = shift(@{$self->{level_rollup_state}->{pending_levels}->{$pending_levels[0]}});
    $self->{level_rollup_state}->{current_level} = $pending_levels[0];
    return $self;
}

sub can_rollup_by_level {
    my($self) = @_;
    return $self->{level_rollup_state};
}



# user help funtions

sub describe {
    my($self) = @_;
    my $cubes = $self->{cube_store}->cubes;
    for(sort { length($a) <=> length($b) || $a cmp $b} keys %$cubes){
        $cubes->{$_}->describe($self);
    }
}

sub get_base_cube_name {
    my($self) = @_;
    return $self->{controller}->{cube_stats}->{base_cube_name};
}

sub base_cube_name {
    my($self) = @_;
    return $self->{controller}->{cube_stats}->{base_cube_name};
}

sub cube_store {
    my($self) = @_;
    return $self->{cube_store};
}

sub cube_list {
    my($self) = @_;
    $self->cube_store->cube_names;
}

sub cube_names {
    my($self) = @_;
    $self->cube_store->cube_names;
}

sub cubes {
    my($self) = @_;
    my @names     = $self->cube_store->cube_names;
    my $cube_hash = $self->cube_store->cubes;
    return map { $cube_hash->{$_} } @names;
}

sub table_store {
    my($self) = @_;
    return $self->{cube_store};
}

sub base_cube {
    my($self) = @_;
    return $self->{base_cube};
}

sub base_table {
    my($self) = @_;
    return $self->{base_cube};
}

sub schema {
    my($self) = @_;
    return $self->{meta_data}->{system}->{base_schema};
}

sub tables {
    my($self) = @_;
    return sort keys %{ $self->table_store->tables };
}

sub table_count {
    my($self) = @_;
    return scalar( keys %{ $self->cube_store->cubes } )
}

sub dmp {
    use Data::Dumper;
    print Dumper( \@_ );
}




1;






__DATA__