DbiResultAgg.cxx

Go to the documentation of this file.

// $Id: DbiResultAgg.cxx,v 1.29 2008/05/12 09:59:11 nwest Exp $


#include <algorithm>
#include <map>
#include <vector>

#include "DatabaseInterface/DbiCache.h"
#include "DatabaseInterface/DbiBinaryFile.h"
#include "DatabaseInterface/DbiDBProxy.h"
#include "DatabaseInterface/DbiResultAgg.h"
#include "DatabaseInterface/DbiResultNonAgg.h"
#include "DatabaseInterface/DbiResultKey.h"
#include "DatabaseInterface/DbiResultSet.h"
#include "DatabaseInterface/DbiTableRow.h"
#include "DatabaseInterface/DbiTimerManager.h"
#include "DatabaseInterface/DbiValidityRecBuilder.h"
#include "MessageService/MsgService.h"
#include "Validity/VldRange.h"
#include "Util/UtilString.h"

ClassImp(DbiResultAgg)

typedef vector<const DbiResult*>::const_iterator ConstResultItr_t;


//   Definition of static data members
//   *********************************

CVSID("$Id: DbiResultAgg.cxx,v 1.29 2008/05/12 09:59:11 nwest Exp $");


//    Definition of all member functions (static or otherwise)
//    *******************************************************
//
//    -  ordered: ctors, dtor, operators then in alphabetical order.

//.....................................................................

DbiResultAgg::DbiResultAgg(const string& tableName,
                           const DbiTableRow* tableRow,
                           DbiCache* cache,
                           const DbiValidityRecBuilder* vrecBuilder,
                           const DbiDBProxy* proxy,
                           const string& sqlQualifiers) :
DbiResult(0,0,sqlQualifiers),
fSize(0)
{
//
//
//  Purpose:  Default constructor
//
//  Arguments: 
//      tableName    in     Name of table.                       
//      tableRow     in     Pointer to a sample tableRow object. 
//                          May be null.
//      cache        in/out Pointer to a cache. May be null.
//      vrecBuilder  in     Pointer to validity record builder containing
//                          aggregated records from query. May be null
//      proxy         in    Pointer to database proxy. May be null.
//      sqlQualifiers in    Extended Context sql qualifiers
//
//  Return:    n/a
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Create Result from DbiResultSet generated by query.


//  Program Notes:-
//  =============

//  tableRow is just used to create new subclass DbiTableRow objects.

  LEA_CTOR    //Leak Checker

  typedef map<UInt_t,UInt_t> seqToRow_t;

  MSG("Dbi", Msg::kVerbose) << "Creating DbiResultAgg" << endl;
  SetTableName(tableName);
  if ( ! tableRow || ! cache || ! vrecBuilder || ! proxy ) return;

// Unpack the extended context SQL qualifiers.
// Don't use StringTok - it eats null strings 
// e.g. abc;;def gives 2 substrings.

  string::size_type loc  = sqlQualifiers.find(';');
  string::size_type loc2 = sqlQualifiers.find(';',loc+1);
  string sqlData  = string(sqlQualifiers,loc+1,loc2-loc-1);
  string fillOpts = string(sqlQualifiers,loc2+1);

//Loop over all rows looking to see if they are already in
//the cache, and if not, recording their associated sequence numbers

  vector<UInt_t> reqSeqNos;  // Sequence numbers required from DB.
  seqToRow_t seqToRow;       // Map SeqNo - > RowNo.
//  Set up a Default database number, it will be updated if anything
//  needs to be read from the database.
  UInt_t dbNo = 0;
  Int_t maxRowNo = vrecBuilder->GetNumValidityRec() - 1;

//Ignore the first entry from the validity rec builder, it will be
//for Agg No = -1, which should not be present for aggregated data.
  for ( Int_t rowNo = 1; rowNo <= maxRowNo; ++rowNo ) {
    const DbiValidityRec& vrecRow = vrecBuilder->GetValidityRec(rowNo);

//  If its already in the cache, then just connect it in.
    const DbiResult* res = cache->Search(vrecRow,sqlQualifiers);
    MSG("Dbi", Msg::kVerbose) << "Checking validity rec " << rowNo
                              << " " << vrecRow
                              << "SQL qual: " << sqlQualifiers
                              << " cache search: " << (void*) res << endl;
    if ( res ) {
      fResults.push_back(res);
      res->Connect();
      fSize += res->GetNumRows();
    }

//  If its not in the cache, but represents a gap, then create an empty
//  DbiResult and add it to the cache.
    else if ( vrecRow.IsGap() ) {
      DbiResult* newRes = new DbiResultNonAgg(0, tableRow, &vrecRow);
      cache->Adopt(newRes,false);
      fResults.push_back(newRes);
      newRes->Connect();
    }

//  Neither in cache, nor a gap, so record its sequence number.
    else {
      UInt_t seqNo = vrecRow.GetSeqNo();
      reqSeqNos.push_back(seqNo);
      seqToRow[seqNo] = rowNo;
      fResults.push_back(0);
//    All data must come from a single database, so any vrec will
//    do to define which one.
      dbNo = vrecRow.GetDbNo();
    }
  }

//If there are required sequence numbers, then read them from the
//database and build DbiResults for each.

  if ( reqSeqNos.size() ) {
//  Sort into ascending order; it may simplify the query which will
//  block ranges of sequence numbers together.
    sort(reqSeqNos.begin(),reqSeqNos.end());
    DbiResultSet* rs = proxy->QuerySeqNos(reqSeqNos,dbNo,sqlData,fillOpts);
//  Flag that data was read from Database.
    this->SetResultsFromDb();
    DbiTimerManager::gTimerManager.StartSubWatch(1);
    while ( ! rs->IsExhausted() ) {
      Int_t seqNo;
      *rs >> seqNo;
      rs->DecrementCurCol();
      Int_t rowNo = -2;
      if ( seqToRow.find(seqNo) == seqToRow.end() ) {
        MSG("Dbi", Msg::kError)
          << "Unexpected SeqNo: " << seqNo << endl;
      }
      else {
        rowNo = seqToRow[seqNo];
        MSG("Dbi", Msg::kVerbose)
          << "Procesing SeqNo: " << seqNo 
          << " for row " << rowNo << endl;
      }

      const DbiValidityRec& vrecRow = vrecBuilder->GetValidityRec(rowNo);
      DbiResultNonAgg* newRes = new DbiResultNonAgg(rs,tableRow,&vrecRow);
//    Don't allow results from Extended Context queries to be reused.
      if ( this->IsExtendedContext() ) newRes->SetCanReuse(false);
      if ( rowNo == -2 ) {
        delete newRes;
      }
      else {
        MSG("Dbi", Msg::kVerbose)
          << "SeqNo: " << seqNo 
          << " produced " << newRes->GetNumRows() << " rows" << endl;
//      Adopt but don't register key for this component, only the overall DbiResultAgg
//      will have a registered key.
        cache->Adopt(newRes,false);
        fResults[rowNo-1] = newRes;
        newRes->Connect();
        fSize += newRes->GetNumRows();
      }
    }

//  DbiResultSet fully processed, so delete it.
    delete rs;
  }

//All component DbiResultNonAgg objects have now been located and
//connected in, so set up their access keys and determine the validty
//range by ANDing the time windows together.

  fRowKeys.reserve(fSize); 

  DbiValidityRec vRec = vrecBuilder->GetValidityRec(1);
  for ( Int_t rowNo = 1; rowNo <= maxRowNo; ++rowNo ) {

    const DbiValidityRec& vrecRow = vrecBuilder->GetValidityRec(rowNo);
    VldRange r = vrecRow.GetVldRange();
    vRec.AndTimeWindow(r.GetTimeStart(),r.GetTimeEnd());

    const DbiResult* res = fResults[rowNo-1];
    if ( res ) {
      UInt_t numEnt = res->GetNumRows();
      for (UInt_t entNo = 0; entNo < numEnt; ++entNo ) 
         fRowKeys.push_back(res->GetTableRow(entNo));
    }
  }

// Now that the row look-up table has been built the natural index
// look-up table can be filled in.
  this->BuildLookUpTable();

// Set aggregate number to -1 to show that it has multiple aggregates
  vRec.SetAggregateNo(-1);
  SetValidityRec(vRec);

  MSG("Dbi",Msg::kDebug) 
    << "Aggregate contains " << fSize  << " entries.  vRec:-" << endl
    << vRec << endl;

   MSG("Dbi",Msg::kSynopsis) << "Created aggregated result set no. of rows: " 
                             << this->GetNumRows() << endl;

}

//.....................................................................

DbiResultAgg::~DbiResultAgg() {
//
//
//  Purpose: Destructor
//
//  Arguments: 
//    None.
//
//  Return:    n/a
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o  Destroy ResultAgg and disconnect all associated DbiResults,


//  Program Notes:-
//  =============

//  None.

  LEA_DTOR    //Leak Checker

  MSG("Dbi", Msg::kVerbose) << "Destroying DbiResultAgg."  << endl;

  for ( ConstResultItr_t itr = fResults.begin();
        itr != fResults.end();
        ++itr) if ( *itr ) (*itr)->Disconnect();

}
//.....................................................................

DbiResultKey* DbiResultAgg::CreateKey() const {
//
//
//  Purpose:  Create a key that corresponds to this result.
//            Caller must take ownership.

  DbiResultKey* key = 0;
  for ( ConstResultItr_t itr = fResults.begin();
        itr != fResults.end();
        ++itr ) {
    const DbiResult* result = *itr;
    if ( result ) {
      // Create key from first result.
      if ( ! key ) key = result->CreateKey();
      // Extend key from the remainder.
      else {
        const DbiValidityRec& vrec = result->GetValidityRec();
        key->AddVRecKey(vrec.GetSeqNo(),vrec.GetCreationDate());
      }
    }
  }

// Should not have an empty set, but just in case.
  if ( ! key ) key = new DbiResultKey();

  return key;

}
//.....................................................................

const DbiTableRow* DbiResultAgg::GetTableRow(UInt_t row) const {
//
//
//  Purpose:  Return TableRow from last query.
//
//  Arguments: 
//    row         in    Required row.
//
//  Return:    TableRow ptr, or =0 if no row.
//
//  Contact:   N. West

//  Program Notes:-
//  =============

//  None.

  return  ( row >= fRowKeys.size() ) ? 0 : fRowKeys[row];

}

//.....................................................................

const DbiValidityRec& DbiResultAgg::GetValidityRec(
                                  const DbiTableRow* row) const {
//
//
//  Purpose:  Get DbiValidityRec associated with table or row.
//
//  Arguments: 
//    row          in    The table row
//                       or null (default) to get table DbiValidityRec
//
//  Contact:   N. West
//

//  Program Notes:-
//  =============

//  The DbiValidityRec for the in-memory table is effectively an 
//  AND of the DbiValidityRecs of the individual rows.

 if ( ! row ) return this->GetValidityRecGlobal();
 DbiResult* owner = row->GetOwner();
 return owner ? owner->GetValidityRecGlobal() : this->GetValidityRecGlobal();

}
//.....................................................................

Bool_t DbiResultAgg::Satisfies(const string& sqlQualifiers)  {
//
//
//  Purpose:  Return true if result satisfies extended context query.
//

  MSG("Dbi",Msg::kDebug) 
    << "Trying to satisfy: SQL: " << sqlQualifiers
    << "\n with CanReuse: " << this->CanReuse() 
    << " sqlQualifiers: " << this->GetSqlQualifiers() 
    << endl;
 return    this->CanReuse()
        && this->GetSqlQualifiers() == sqlQualifiers;
}

//.....................................................................

void DbiResultAgg::Streamer(DbiBinaryFile& bf) {
//
//
//  Purpose:  I/O to binary file
//
//  Specification:-
//  =============

//  Output constituent non-gap DbiResultNonAgg objects.

  vector<const DbiResult*>::const_iterator itr = fResults.begin();
  vector<const DbiResult*>::const_iterator end = fResults.end();

  UInt_t numNonAgg = 0;
  for (; itr != end; ++itr) {
    const DbiResultNonAgg* rna = dynamic_cast<const DbiResultNonAgg*>(*itr);
    if ( rna && ! rna->GetValidityRecGlobal().IsGap() ) ++numNonAgg;
  }
  bf << numNonAgg;

  
  for (itr = fResults.begin(); itr != end; ++itr) {
    const DbiResultNonAgg* rna = dynamic_cast<const DbiResultNonAgg*>(*itr);
    if ( rna && ! rna->GetValidityRecGlobal().IsGap() ) bf << *rna;
  }
}

/*    Template for New Member Function

//.....................................................................

DbiResultAgg:: {
//
//
//  Purpose:  
//
//  Arguments: 
//    xxxxxxxxx    in    yyyyyy
//
//  Return:    
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o 

//  Program Notes:-
//  =============

//  None.


}

*/

---