package Common::DB::Export::Parallel;

use constant 'THREADS'   => 0;
use constant 'CHUNKSIZE' => 1_000_000;

use strict;
use warnings;

use Parallel::ForkManager;

use lib '/app/tools/common/lib';
use Common::Session;

use base 'Common::DB::Export';

__PACKAGE__->mk_accessors(qw/database compress chunksize parallel/);

sub _init {
    my $self = shift;
    $self->compress(1);
    $self->chunksize(CHUNKSIZE);
    $self->parallel(THREADS);
    $self->_init_db if not ref $self->database;
}

sub export_table {
    my $self  = shift;
    my $table = shift;
    my $sql   = shift;

    ## get the PRIMARY key from the table to get the max id
    my $dbx    = Common::Session::getdbx('localhost');
    my $schema = $self->database->db_name;
    my $search = { 'table_schema' => $schema, 'table_name' => $table, 'index_name' => 'PRIMARY' };
    my $idx    = $dbx->resultset('InformationSchema::Indexes')->search($search);
    if ( not defined $idx or $idx->count == 0 ) {
        logMessage( 'error', "No primary key found. Is $schema on this server with PKs?" );
        return 0;
    } elsif ( $idx->count > 1 ) {
        logMessage( 'error', "Mutliple primary key tables not support for parallel export" );
    	return 0;    
    }

    ## get the maxid from primary key - break up the table in chunks using max
    my $pk     = $idx->first;
    my $dbh    = $dbx->getdbh;
    my $maxsql = "SELECT MAX(" . $pk->column_name . ") FROM ${schema}.${table}";
    my $max    = $dbh->selectcol_arrayref($maxsql)->[0];
    if ( not defined $max or $max < 1 ) {
        logMessage( 'warn', 'Empty table not exporting' );
        return 0;
    }
    my $chunks = int( $max / $self->chunksize ) + 1;

    undef $idx;
    $dbx->getdbh->disconnect;

    ## query chunks with parallel threads
    my $pm = Parallel::ForkManager->new( $self->parallel );

    my $pnum = 1;
    my $i    = 1;
    while ( $i <= ( $chunks * $self->chunksize ) ) {

        ## set min/max
        my $min = $i;
        my $max = $self->chunksize + $i;

        ## min to max
        $i = $max;

        ## file partiton number
        my $part = sprintf( "%04d", $pnum++ );

        $pm->start and next;

        ## need a new db connection for every thread
        my $dbh = Common::Session::getdbh('localhost');

        my $path   = $self->directory->localcreate;
        my $file   = "$path/${table}.${part}.txt";
        my $logidx = sprintf( "(%04d/%04d)", $part, $chunks );
        logMessage( 'info2', "$logidx $schema - partition:$part min:$min max:$max" );

        my $sql = defined $sql ? $sql : $self->_sql;
        $sql = $self->_writesql( $sql, $file, $table, $pk->column_name, $min, $max, $logidx );
        $dbh->do($sql);
        if ( $dbh->err ) {
            logMessage( 'error2', $dbh->errstr );
            exit 1;
        }
        $dbh->disconnect;

        $pm->finish;

    }
    logMessage( 'info2', "waiting for exports to finish" );
    $pm->wait_all_children;
    logMessage( 'info', "COPY complete" );

    if ( $self->compress ) {
        my $pmc = Parallel::ForkManager->new( $self->parallel );
        foreach my $file ( @{ $self->directory->localfiles("*.txt") } ) {
            $pmc->start and next;
            $file->compress;
            $pmc->finish;
        }
        $pmc->wait_all_children;
    }

    [ $self->directory->localfiles ];

}

sub _sql {
    my $self = shift;
    my $sql  = (
        qq{
SELECT /* :comment: */ * 
FROM :database:.:table:
WHERE 1
AND :primary: >= :min:
AND :primary:  < :max:
INTO OUTFILE ":file:"
FIELDS TERMINATED BY '\t' OPTIONALLY ENCLOSED BY '"'
	}
    );
    $sql;
}

sub _writesql {
    my $self    = shift;
    my $sql     = shift;
    my $file    = shift;
    my $table   = shift;
    my $primary = shift;
    my $min     = shift;
    my $max     = shift;
    my $comment = shift;

    my $schema = $self->database->db_name;

    $sql =~ s/:database:/$schema/smg;
    $sql =~ s/:file:/$file/smg;
    $sql =~ s/:table:/$table/smg;
    $sql =~ s/:primary:/$primary/smg;
    $sql =~ s/:max:/$max/smg;
    $sql =~ s/:min:/$min/smg;
    $sql =~ s/:comment:/$comment/smg if defined $comment;

    $sql;

}

1;