Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor

  $analysisJobAdaptor = $db_adaptor->get_AnalysisJobAdaptor;
  $analysisJobAdaptor = $analysisJob->adaptor;

  Module to encapsulate all db access for persistent class AnalysisJob.
  There should be just one per application and database connection.

  my ($class, @args) = @_;

  return undef unless(scalar @args);

  my ($input_id, $analysis, $prev_analysis_job_id, $blocked) =
     rearrange([qw(INPUT_ID ANALYSIS input_job_id BLOCK )], @args);

  $prev_analysis_job_id=0 unless($prev_analysis_job_id);
  throw("must define input_id") unless($input_id);
  throw("must define analysis") unless($analysis);
  throw("analysis must be [Bio::EnsEMBL::Analysis] not a [$analysis]")
    unless($analysis->isa('Bio::EnsEMBL::Analysis'));
  throw("analysis must have adaptor connected to database")
    unless($analysis->adaptor and $analysis->adaptor->db);

  if(length($input_id) >= 255) {
    my $input_data_id = $analysis->adaptor->db->get_AnalysisDataAdaptor->store_if_needed($input_id);
    $input_id = "_ext_input_analysis_data_id $input_data_id";
  }

  my $sql = q{INSERT ignore into analysis_job 
              (input_id, prev_analysis_job_id,analysis_id,status)
              VALUES (?,?,?,?)};
 
  my $status ='READY';
  $status = 'BLOCKED' if($blocked);

  my $dbc = $analysis->adaptor->db->dbc;
  my $sth = $dbc->prepare($sql);
  $sth->execute($input_id, $prev_analysis_job_id, $analysis->dbID, $status);
  my $dbID = $sth->{'mysql_insertid'};
  $sth->finish;

  $dbc->do("UPDATE analysis_stats SET ".
           "total_job_count=total_job_count+1 ".
           ",unclaimed_job_count=unclaimed_job_count+1 ".
           ",status='LOADING' ".
           "WHERE status!='BLOCKED' and analysis_id='".$analysis->dbID ."'");

  return $dbID;
}

###############################################################################
#
#  INSTANCE methods
#
###############################################################################

  my $self = shift;

  return qw (a.analysis_job_id  
             a.prev_analysis_job_id
             a.analysis_id	      
             a.input_id 
             a.job_claim  
             a.hive_id	      
             a.status 
             a.retry_count          
             a.completed
             a.branch_code
             a.runtime_msec
             a.query_count
            );

  my $self = shift;
  return '';

  my $self = shift;
  return 'ORDER BY retry_count';

  my ($self, $constraint, $join) = @_;
  
  my @tables = $self->_tables;
  my $columns = join(', ', $self->_columns());
  
  if ($join) {
    foreach my $single_join (@{$join}) {
      my ($tablename, $condition, $extra_columns) = @{$single_join};
      if ($tablename && $condition) {
        push @tables, $tablename;
        
        if($constraint) {
          $constraint .= " AND $condition";
        } else {
          $constraint = " $condition";
        }
      } 
      if ($extra_columns) {
        $columns .= ", " . join(', ', @{$extra_columns});
      }
    }
  }
      
  #construct a nice table string like 'table1 t1, table2 t2'
  my $tablenames = join(', ', map({ join(' ', @$_) } @tables));

  my $sql = "SELECT $columns FROM $tablenames";

  my $default_where = $self->_default_where_clause;
  my $final_clause = $self->_final_clause;

  #append a where clause if it was defined
  if($constraint) { 
    $sql .= " WHERE $constraint ";
    if($default_where) {
      $sql .= " AND $default_where ";
    }
  } elsif($default_where) {
    $sql .= " WHERE $default_where ";
  }

  #append additional clauses which may have been defined
  $sql .= " $final_clause";

  my $sth = $self->prepare($sql);
  $sth->execute;  

  #print STDOUT $sql,"\n";

  return $self->_objs_from_sth($sth);

  my ($self, $sth) = @_;
  
  my %column;
  $sth->bind_columns(\(  @column{ @{$sth->{NAME_lc} } } ));

  my @jobs = ();
    
  while ($sth->fetch()) {
    my $job = new Bio::EnsEMBL::Hive::AnalysisJob;

    $job->dbID($column{'analysis_job_id'});
    $job->analysis_id($column{'analysis_id'});
    $job->input_id($column{'input_id'});
    $job->job_claim($column{'job_claim'});
    $job->hive_id($column{'hive_id'});
    $job->status($column{'status'});
    $job->retry_count($column{'retry_count'});
    $job->completed($column{'completed'});
    $job->branch_code($column{'branch_code'});
    $job->runtime_msec($column{'runtime_msec'});
    $job->query_count($column{'query_count'});
    $job->adaptor($self);
    
    if($column{'input_id'} =~ /_ext_input_analysis_data_id (\d+)/) {
      #print("input_id was too big so stored in analysis_data table as dbID $1 -- fetching now\n");
      $job->input_id($self->db->get_AnalysisDataAdaptor->fetch_by_dbID($1));
    }

    push @jobs, $job;    
  }
  $sth->finish;
  
  return\@ jobs
}


#
# STORE / UPDATE METHODS
#
################

  my $self = shift;

  return (['analysis_job', 'a']);

  my $self     = shift;
  my $worker   = shift;

  throw("must define worker") unless($worker);

  my $ug    = new Data::UUID;
  my $uuid  = $ug->create();
  my $claim = $ug->to_string( $uuid );
  #print("claiming jobs for hive_id=", $worker->hive_id, " with uuid $claim\n");
  my $status = 'READY';
  $status = 'HIGHMEM' if (defined($worker->{HIGHMEM}));
  my $sql_base = "UPDATE analysis_job SET job_claim='$claim'".
                 " , hive_id='". $worker->hive_id ."'".
                 " , status='CLAIMED'".
                 " WHERE job_claim='' and status='" . $status . "'". 
                 " AND analysis_id='" .$worker->analysis->dbID. "'"; 

  my $sql_virgin = $sql_base .  
                   " AND retry_count=0".
                   " LIMIT " . $worker->batch_size;

  my $sql_any = $sql_base .  
                " LIMIT " . $worker->batch_size;
  
  my $claim_count = $self->dbc->do($sql_virgin);
  if($claim_count == 0) {
    $claim_count = $self->dbc->do($sql_any);
  }
  return $claim;

  my $self = shift;

  return $self->_generic_fetch();

  my ($self,$analysis_id) = @_;

  my $constraint = "a.status='FAILED'";
  $constraint .= " AND a.analysis_id=$analysis_id" if($analysis_id);
  return $self->_generic_fetch($constraint);

  my ($self,$claim,$analysis_id) = @_;

  throw("fetch_by_claim_analysis must have claim ID") unless($claim);
  throw("fetch_by_claim_analysis must have analysis_id") unless($analysis_id);
  my $constraint = "a.status='CLAIMED' AND a.job_claim='$claim' AND a.analysis_id='$analysis_id'";
  return $self->_generic_fetch($constraint);

  my ($self,$id) = @_;

  unless(defined $id) {
    throw("fetch_by_dbID must have an id");
  }

  my @tabs = $self->_tables;

  my ($name, $syn) = @{$tabs[0]};

  #construct a constraint like 't1.table1_id = 1'
  my $constraint = "${syn}.${name}_id = $id";

  #return first element of _generic_fetch list
  my ($obj) = @{$self->_generic_fetch($constraint)};
  return $obj;

  my ($self,$hive_id,$analysis_id) = @_;

  throw("fetch_by_run_analysis must have hive_id") unless($hive_id);
  throw("fetch_by_run_analysis must have analysis_id") unless($analysis_id);
  my $constraint = "a.status='RUN' AND a.hive_id=$hive_id AND a.analysis_id='$analysis_id'";
  return $self->_generic_fetch($constraint);

  my $self = shift;
  my $query = shift;

  return undef unless($query);
  #print("Bio::EnsEMBL::DBSQL::AnalysisAdaptor::fetch_by_url_query : $query\n");

  if((my $p=index($query, "=")) != -1) {
    my $type = substr($query,0, $p);
    my $value = substr($query,$p+1,length($query));

    if($type eq 'dbID') {
      return $self->fetch_by_dbID($value);
    }
  }
  return undef;
}

#
# INTERNAL METHODS
#
###################

  my $self   = shift;
  my $job    = shift;

  my $ug    = new Data::UUID;
  my $uuid  = $ug->create();
  $job->job_claim($ug->to_string( $uuid ));

  my $sql = "UPDATE analysis_job SET status='CLAIMED', job_claim=?, hive_id=? WHERE analysis_job_id=?";

  #print("$sql\n");            
  my $sth = $self->prepare($sql);
  $sth->execute($job->job_claim, $job->hive_id, $job->dbID);
  $sth->finish;

  my $self        = shift;
  my $analysis_id = shift;

  throw("must define analysis_id") unless($analysis_id);

  my $sql;
  #first just reset the claimed jobs, these don't need a retry_count index increment
  $sql = "DELETE FROM analysis_stats WHERE analysis_id=$analysis_id";
  $self->dbc->do($sql);
  $sql = "ANALYZE TABLE analysis_stats";
  $self->dbc->do($sql);
  $sql = "DELETE FROM analysis_job WHERE analysis_id=$analysis_id";
  $self->dbc->do($sql);
  $sql = "ANALYZE TABLE analysis_job";
  $self->dbc->do($sql);
  $sql = "DELETE FROM hive WHERE analysis_id=$analysis_id";
  $self->dbc->do($sql);
  $sql = "ANALYZE TABLE hive";
  $self->dbc->do($sql);

}


1;

  my $self        = shift;
  my $analysis_id = shift;

  throw("must define analysis_id") unless($analysis_id);

  my ($sql, $sth);
  $sql = "UPDATE analysis_job SET job_claim='', status='READY' WHERE status!='BLOCKED' and analysis_id=?";
  $sth = $self->prepare($sql);
  $sth->execute($analysis_id);
  $sth->finish;

  $self->db->get_AnalysisStatsAdaptor->update_status($analysis_id, 'LOADING');

  my $self = shift;
  my $job_id = shift;

  #added hive_id index to analysis_job table which made this operation much faster

  my $sql;
  #first just reset the claimed jobs, these don't need a retry_count index increment
  $sql = "UPDATE analysis_job SET job_claim='', status='READY'".
         " WHERE status='CLAIMED'".
         " AND analysis_job_id=$job_id";
  $self->dbc->do($sql);
  #print("  done update CLAIMED\n");

  # an update with select on status and hive_id took 4seconds per worker to complete,
  # while doing a select followed by update on analysis_job_id returned almost instantly
  
  $sql = "
    UPDATE analysis_job, analysis_stats
    SET job_claim='', analysis_job.status='READY', retry_count=retry_count+1
    WHERE
      analysis_job.status in ('GET_INPUT','RUN','WRITE_OUTPUT')
      AND analysis_job.analysis_id = analysis_stats.analysis_id
      AND retry_count < max_retry_count
      AND analysis_job_id=$job_id";
  #print("$sql\n");
  $self->dbc->do($sql);

  $sql = "
    UPDATE analysis_job, analysis_stats
    SET job_claim='', analysis_job.status='FAILED', retry_count=retry_count+1
    WHERE
      analysis_job.status in ('GET_INPUT','RUN','WRITE_OUTPUT')
      AND analysis_job.analysis_id = analysis_stats.analysis_id
      AND retry_count >= max_retry_count
      AND analysis_job_id=$job_id";
  #print("$sql\n");
  $self->dbc->do($sql);

  #print(" done update BROKEN jobs\n");

  my $self = shift;
  my $worker = shift;
  throw("must define worker") unless($worker);

  #added hive_id index to analysis_job table which made this operation much faster

  my ($sql, $sth);
  my $max_retry_count = $worker->analysis->stats->max_retry_count();
  #first just reset the claimed jobs, these don't need a retry_count index increment
  $sql = "UPDATE analysis_job SET job_claim='', status='READY'".
         " WHERE status='CLAIMED'".
         " AND hive_id='" . $worker->hive_id ."'";
  $sth = $self->prepare($sql);
  $sth->execute();
  $sth->finish;
  #print("  done update CLAIMED\n");

  # an update with select on status and hive_id took 4seconds per worker to complete,
  # while doing a select followed by update on analysis_job_id returned almost instantly
  
  $sql = "UPDATE analysis_job SET job_claim='', status='READY'".
         " ,retry_count=retry_count+1".
         " WHERE status in ('GET_INPUT','RUN','WRITE_OUTPUT')".
	 " AND retry_count<$max_retry_count".
         " AND hive_id='" . $worker->hive_id ."'";
  #print("$sql\n");
  $sth = $self->prepare($sql);
  $sth->execute();
  $sth->finish;

  $sql = "UPDATE analysis_job SET status='FAILED'".
         " ,retry_count=retry_count+1".
         " WHERE status in ('GET_INPUT','RUN','WRITE_OUTPUT')".
	 " AND retry_count>=$max_retry_count".
         " AND hive_id='" . $worker->hive_id ."'";
  #print("$sql\n");
  $sth = $self->prepare($sql);
  $sth->execute();
  $sth->finish;

  #print(" done update BROKEN jobs\n");

  my $self = shift;
  my $job_id = shift;

  #added hive_id index to analysis_job table which made this operation much faster

  my $sql;
  #first just reset the claimed jobs, these don't need a retry_count index increment
  $sql = "UPDATE analysis_job SET job_claim='', status='HIGHMEM'".
         " WHERE analysis_job_id=$job_id";
  $self->dbc->do($sql);
  #print("  done update CLAIMED\n");

  my $self = shift;
  my $analysis_job_id   = shift;
  throw("must define job") unless($analysis_job_id);

  my ($sql, $sth);
  #first just reset the claimed jobs, these don't need a retry_count index increment
  $sql = "UPDATE analysis_job SET hive_id=0, job_claim='', status='READY', retry_count=0 WHERE analysis_job_id=?";
  $sth = $self->prepare($sql);
  $sth->execute($analysis_job_id);
  $sth->finish;
  #print("  done update CLAIMED\n");

  my ($self,$job) = @_;

  return unless($job);

  my $sql = sprintf("DELETE from analysis_job_file WHERE hive_id=%d and analysis_job_id=%d",
                   $job->hive_id, $job->dbID);
  $self->dbc->do($sql);
  return unless($job->stdout_file or $job->stderr_file);

  $sql = "INSERT ignore INTO analysis_job_file (analysis_job_id, hive_id, retry, type, path) VALUES ";
  if($job->stdout_file) {
    $sql .= sprintf("(%d,%d,%d,'STDOUT','%s')", $job->dbID, $job->hive_id, 
		    $job->retry_count, $job->stdout_file); 
  }
  $sql .= "," if($job->stdout_file and $job->stderr_file);
  if($job->stderr_file) {
    $sql .= sprintf("(%d,%d,%d,'STDERR','%s')", $job->dbID, $job->hive_id, 
		    $job->retry_count, $job->stderr_file); 
  }
 
  $self->dbc->do($sql);

  my ($self,$job) = @_;

  my $sql = "UPDATE analysis_job SET status='".$job->status."' ";
  if($job->status eq 'DONE') {
    $sql .= ",completed=now(),branch_code=".$job->branch_code;
    $sql .= ",runtime_msec=".$job->runtime_msec;
    $sql .= ",query_count=".$job->query_count;
  }
  if($job->status eq 'READY') {
    $sql .= ",job_claim=''";
  }
  $sql .= " WHERE analysis_job_id='".$job->dbID."' ";
  
  my $sth = $self->prepare($sql);
  $sth->execute();
  $sth->finish;

  Contact Jessica Severin on implemetation/design detail: jessica@ebi.ac.uk
  Contact Ewan Birney on EnsEMBL in general: birney@sanger.ac.uk

  The rest of the documentation details each of the object methods.
  Internal methods are preceded with a _