DbiSqlValPacket.cxx

Go to the documentation of this file.

// $Id: DbiSqlValPacket.cxx,v 1.43 2007/04/26 14:19:57 west Exp $

#include <algorithm>
#include <memory>
using std::auto_ptr;
#include <sstream>
#include <vector>

#include "DatabaseInterface/Dbi.h"
#include "DatabaseInterface/DbiCascader.h"
#include "DatabaseInterface/DbiConfigSet.h"
#include "DatabaseInterface/DbiOutRowStream.h"
#include "DatabaseInterface/DbiResultPtr.h"
#include "DatabaseInterface/DbiResultSet.h"
#include "DatabaseInterface/DbiSqlValPacket.h"
#include "DatabaseInterface/DbiStatement.h"
#include "DatabaseInterface/DbiTableProxy.h"
#include "DatabaseInterface/DbiTableRow.h"
#include "DatabaseInterface/DbiTableProxyRegistry.h"
#include "DatabaseInterface/DbiValidityRec.h"
#include "DatabaseInterface/DbiValRecSet.h"

#include "LeakChecker/Lea.h"
#include "MessageService/MsgService.h"
#include "Util/UtilString.h"
#include "Validity/VldRange.h"

ClassImp(DbiSqlValPacket)

#ifdef IRIX6
// SGI barfs at generating an operator != for this enum if it
// is anonymous and in the Fill() method.
enum EFillState { kLOOKING_FOR_HEADER,
                  kLOOKING_FOR_TRAILER };
#endif

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

Bool_t DbiSqlValPacket::fgLastMetaHadRowCounter = false;

CVSID("$Id: DbiSqlValPacket.cxx,v 1.43 2007/04/26 14:19:57 west Exp $");

//   Definition of file static members functions
//   *******************************************

static bool compStringPtrs(const string* str1, const string* str2 ) {
  return *str1 < *str2; }

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

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

DbiSqlValPacket::DbiSqlValPacket() :
fNumErrors(0),
fSeqNo(0),
fNumStmts(0)
{
//
//
//  Purpose:  Default ctor.

  LEA_CTOR    //Leak Checker

  MSG("Dbi", Msg::kVerbose) << "Creating DbiSqlValPacket" << endl;
}

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

DbiSqlValPacket::DbiSqlValPacket(std::ifstream& is) :
fNumErrors(0),
fSeqNo(0),
fNumStmts(0)
{
//
//
//  Purpose:  Constructor reading from input stream.
//
//  Arguments:
//      is    in/out    Input stream.
//
//  Return:   n/a. 
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o  Create DbiSqlValPacket and fill from input stream.


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

//  None.

  LEA_CTOR    //Leak Checker

  MSG("Dbi", Msg::kVerbose) << "Creating DbiSqlValPacket" << endl;

  Fill(is);

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

DbiSqlValPacket::DbiSqlValPacket(const DbiValidityRec& vrec) :
fNumErrors(0),
fSeqNo(vrec.GetSeqNo()),
fNumStmts(0),
fTableName(vrec.GetTableProxy()->GetTableName()),
fCreationDate(vrec.GetCreationDate())
{
//
//
//  Purpose:  Constructor from a DbiValidityRec.
//
//  Arguments:
//      vrec  in   DbiValidityRec from which to create packet.     .
//
//  Return:   n/a. 
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o  Create DbiSqlValPacket from DbiValidityRec and associated data.


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

//  This member function uses DbiConfigSet as a generic concrete
//  DbiTableRow that can be used to create SQL for any type of table.

  LEA_CTOR    //Leak Checker

  MSG("Dbi", Msg::kVerbose) << "Creating DbiSqlValPacket" << endl;

  const DbiTableProxy& tableProxy = *vrec.GetTableProxy();
  Int_t seqNo  = vrec.GetSeqNo();
  UInt_t dbNo  = vrec.GetDbNo();

  // Create the SQL for the DbiValidityRec itself.
  this->AddRow(tableProxy,0,vrec);

  // Create the SQL for the rows.

  const DbiDBProxy& dbProxy = tableProxy.GetDBProxy();
  DbiResultSet* rset = dbProxy.QuerySeqNo(seqNo,dbNo);

  
  for(; ! rset->IsExhausted(); rset->FetchRow()) {
    string str;
    rset->RowAsCsv(str);
    this->AddRow(str);
  }
  delete rset;
  rset = 0;
}

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

DbiSqlValPacket::~DbiSqlValPacket() {
//
//
//  Purpose: Destructor
//
//  Arguments: 
//    None.
//
//  Return:    n/a
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o  Destroy DbiSqlValPacket.


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

//  None.

  LEA_DTOR    //Leak Checker

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

  Clear();

}

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

Bool_t DbiSqlValPacket::AddDataRow(const DbiTableProxy& tblProxy,
                                   const DbiValidityRec* vrec,
                                   const DbiTableRow& row){
//
//
//  Purpose: Add data row. 
//

  if ( this->GetNumSqlStmts() == 0 ) {
    MAXMSG("Dbi",Msg::kError,20) 
      << "Cannot add data row - packet does not have a VLD row"  << endl;
    ++fNumErrors;
    return kFALSE;
  }

  return this->AddRow(tblProxy,vrec,row);

}

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

void DbiSqlValPacket::AddRow(const string & row){
//
//
//  Purpose: Add row. 
//

  string sql("INSERT INTO ");
  sql += this->GetTableName();
  if ( this->GetNumSqlStmts() == 0 ) sql += "VLD";
  sql += " VALUES (" + row + ");";
  ostringstream seqno;
  seqno << this->GetSeqNo();
  this->SetSeqNoOnRow(sql,seqno.str());
  fSqlStmts.push_back(sql);
  ++fNumStmts;

}

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

Bool_t DbiSqlValPacket::AddRow(const DbiTableProxy& tblProxy,
                               const DbiValidityRec* vrec,
                               const DbiTableRow& row){
//
//
//  Purpose: Add row. 
//


  bool isVld = this->GetNumSqlStmts() == 0;
  const DbiTableMetaData& meta = isVld ? tblProxy.GetMetaValid() : tblProxy.GetMetaData(); 
  DbiOutRowStream outRow(&meta);

// Store dummy SEQNO, and, if necessary ROW_COUNTER, for data rows. 
  if ( ! isVld ) {
    outRow << 0;  // dummy SEQNO
    if ( outRow.HasRowCounter() )  outRow <<  fNumStmts; // ROW_COUNTER
  }
  row.Store(outRow,vrec);
  if ( ! outRow.HasGoodData() ) {
    if ( ! outRow.IsComplete() ) {
      MAXMSG("Dbi",Msg::kError,20)
        << "Incomplete data supplied for row " << this->GetNumSqlStmts()-1
        << " of table " 
        << tblProxy.GetTableName() << endl;
    }
    ++fNumErrors;
    return kFALSE;
  }
  this->AddRow(outRow.GetCSV());
  return kTRUE;
}
//.....................................................................

DbiSqlValPacket::CompResult_t DbiSqlValPacket::Compare(
                                const DbiSqlValPacket& that,
                                Bool_t log,
                                const Char_t* thisName,
                                const Char_t* thatName ) const {
//
//
//  Purpose:  Compare to another DbiSqlValPacket
//
//  Arguments: 
//    that         in    The other DbiSqlValPacket to be compared.
//    log          in    If true list differences to MSG("Dbi",kInfo)
//    thisName     in    Optional name for this packet (default: this)
//    thatName     in    Optional name for that packet (default: that)
//
//  Return:    kIdentical   if identical (apart from InsertDate)
//             kUpdate      "that" is more up to date
//             kOutOfDate   "that" is out of date
//             kConflict    records are incompatible
//
//  Contact:   N. West
//

  if ( this->IsEqual(that,log,thisName,thatName ) ) return kIdentical;

  if (    fSeqNo           != that.fSeqNo
       || fTableName       != that.fTableName     ) return kConflict;

  std::vector<std::string> valuesThis = this->GetStmtValues(0);
  std::vector<std::string> valuesThat = that.GetStmtValues(0);

  // Assume CreationDate is the 7th element (counting from 0).
  int comp = valuesThis[7].compare(valuesThat[7]);

  if ( comp < 0 ) {
    if ( log ) MSG("Dbi",Msg::kInfo) 
      << "  Difference classified as Update" << endl;;
    return kUpdate;
  }
  else if ( comp > 0 ) {
    if ( log ) MSG("Dbi",Msg::kInfo) 
      << "  Difference classified as OutOfDate" << endl;
    return kOutOfDate;
  }
  return kConflict;

}


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

Bool_t DbiSqlValPacket::CreateTable(UInt_t dbNo) const {
//
//
//  Purpose:  Create table in specified database.
//
//  Arguments: 
//    dbNo         in    Number of database in the cascade.
//
//  Return:    kTRUE if successfully created.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o If SQL to create tables is available, use it to create 
//    the main and auxiliary tables and refresh the corresponding 
//    DbiTableProxy.

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

//  None.

  if ( ! CanBeStored() ) return kFALSE; 

  // Locate required DbiStatement.
  auto_ptr<DbiStatement> stmtDb(DbiTableProxyRegistry::Instance()
                               .GetCascader()
                               .CreateStatement(dbNo));
  if ( ! stmtDb.get() ) {
    MSG("Dbi",Msg::kWarning) 
      << "Attempting to write to non-existant cascade entry " << dbNo
      << endl;
    return kFALSE;
  }
  if ( fSqlMySqlMetaMain == "" || fSqlMySqlMetaVld  == "" ) {
    MSG("Dbi",Msg::kWarning) 
      << "No SQL available to create table " << fTableName
      << " in cascade entry: " << dbNo << endl;
    return kFALSE;
  }

  stmtDb->ExecuteUpdate(fSqlMySqlMetaVld.c_str());
  if ( stmtDb->PrintExceptions() ) return kFALSE;
  stmtDb->ExecuteUpdate(fSqlMySqlMetaMain.c_str());
  if ( stmtDb->PrintExceptions() ) return kFALSE;

  DbiTableProxyRegistry::Instance().RefreshMetaData(this->GetTableName());

  return kTRUE;

}

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

Bool_t DbiSqlValPacket::Fill(std::ifstream& is) {
//
//
//  Purpose:  Refill object from input string.
//
//  Arguments: 
//      is    in/out    Input stream.
//
//  Return:   kTRUE is object has been refilled. 
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Collect SQL statements, and possibly metadata (as SQL to
//    create table) bracketed by >>>> ... <<<<< checking that
//    the table name and SeqNos match.  Count and discard any data that
//    looks bad. 

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

//  None.

#ifdef IRIX6
  EFillState state = kLOOKING_FOR_HEADER;
#else
  enum { kLOOKING_FOR_HEADER,
         kLOOKING_FOR_TRAILER } state = kLOOKING_FOR_HEADER;
#endif

  enum { kMAXTABLENAMELEN        = Dbi::kMAXTABLENAMELEN,
         kHEADER_TRAILER_MAX_LEN = kMAXTABLENAMELEN + 20 };

  string nameHead;
  string nameTrail;
  UInt_t seqNoHead  = 0;
  UInt_t seqNoTrail = 0;

  string line;
  string msg;
  string sql;
  int lineNum = 0;

  this->Reset();
  //  Loop until EOF reading lines.

  while ( ! is.eof() ) {
    getline(is,line);
    ++lineNum;
    // Skip null lines.
    if (line.size() == 0 ) continue;

    // Look for header line
    if ( state == kLOOKING_FOR_HEADER ) {
      if ( line.substr(0,5) == ">>>>>" ) {
        if ( line.size() >= kHEADER_TRAILER_MAX_LEN ) {
          Report("Bad header",lineNum,line);
          continue;
        }

        // Look for optional metadata.
        if ( line.find("Metadata") != string::npos ) {
          getline(is,fSqlMySqlMetaVld);
          ++lineNum;
          getline(is,fSqlMySqlMetaMain);
          // Record whether or not table has a ROW_COUNTER column.
          fgLastMetaHadRowCounter = 
            fSqlMySqlMetaMain.find("ROW_COUNTER",fSqlMySqlMetaMain.find("(")) != string::npos;
          ++lineNum;
          getline(is,line);
          ++lineNum;
          if (   line.substr(0,5) != "<<<<<" 
              || line.find("Metadata") == string::npos ) {
            Report("Bad metadata",lineNum,line);
            continue;
          }
          getline(is,line);
          ++lineNum;
          if ( line.size() >= kHEADER_TRAILER_MAX_LEN ) {
            Report("Bad header",lineNum,line);
            continue;
          }
        }
          
        //  Collect table name and SeqNo.
        istringstream istr(line.substr(5));
        istr.width(kMAXTABLENAMELEN);
        istr >> nameHead >> seqNoHead;
        if ( ! istr.eof() ) {
          Report("Input error",lineNum,line);
          continue;
        }

        // Header looks good, start to collect SQL. 
        state = kLOOKING_FOR_TRAILER;
        sql = "";
      }
      else {
        Report("Not header",lineNum,line);
      }
    }

    //Look for trailer line.

    else {
      if ( line.substr(0,5) == "<<<<<" ) {
        if ( line.size() >= kHEADER_TRAILER_MAX_LEN 
           ) msg = "Bad trailer";

        else {

          //  Collect table name and SeqNo.
          istringstream istr(line.substr(5));
          istr.width(kMAXTABLENAMELEN);
          istr >> nameTrail >> seqNoTrail;
          if ( ! istr.eof() ) msg = "Input error";

          else if (    nameTrail != nameHead
                    || seqNoHead != seqNoTrail ) {
            msg = "Header/Trailer mismatch: Header: ";
            msg += istr.str();
          }
          else if ( GetNumSqlStmts() == 0 
                  ) msg = "No SQL statements between Header/Trailer";
          else {

            // Trailer looks good return with object filled. 
            fSeqNo     = seqNoHead;
            fTableName = nameHead;

            //Dig out the creation date from the first record.
            string date = this->GetStmtValues(0)[7];
            //Remove the quotes.
            date.erase(0,1);
            date.erase(date.size()-1,1);
            fCreationDate = Dbi::MakeTimeStamp(date);
            // Temporary hack: set for Agg No = -1 for DBUSUBRUNSUMMARY
            if ( fTableName  == "DBUSUBRUNSUMMARY" ) {
              std::vector<std::string> ls;
              UtilString::StringTok(ls,*fSqlStmts.begin(),",");
              // Agg. no is element 6.
              if (ls.size() >= 7 && ls[6] != "-1") {
                static bool warnOnce = true;
                if ( warnOnce ) {
                  MSG("Dbi",Msg::kWarning)
                   << "Setting aggregate number for DBUSUBRUNSUMMARY = -1\n"
                   << "  Expect this message once.  DBUSUBRUNSUMMARY needs to be "
                   << " fixed!" << endl;
                   warnOnce = false;
                }
                ls[6] = "-1";
                sql = ls[0];
                for (unsigned i=1; i<ls.size(); ++i ) sql += "," + ls[i];
                fSqlStmts.pop_front();
                fSqlStmts.push_front(sql);  
              }
            }
            return kTRUE;

          }
        }

        // Trailer bad, start again!
        Report(msg.c_str(),lineNum,line);
        state = kLOOKING_FOR_HEADER;
      }

      // Not a trailer line, must be SQL, collect and append lines
      // until a trailing ; found.
      else {
        sql += line;
        if ( sql[sql.size()-1] == ';') {
          fSqlStmts.push_back(sql);
          ++fNumStmts;
          sql = "";
        }
      } 
    }

  }

  if ( state != kLOOKING_FOR_HEADER 
     ) Report("Unexpected EOF",lineNum,"EOF");
  this->Reset();
  return kFALSE;

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

string DbiSqlValPacket::GetStmt(UInt_t stmtNo) const {
//
//
//  Purpose:  Return a selected statment

  if ( stmtNo >= this->GetNumSqlStmts() ) return "";

  // Locate statement
  std::list<std::string>::const_iterator itr = fSqlStmts.begin();
  while ( stmtNo ) { ++itr; --stmtNo; }

  return *itr;

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

std::vector<std::string> DbiSqlValPacket::GetStmtValues(UInt_t stmtNo) const {
//
//
//  Purpose:  Return all the values associated with a selected statment
//
//  Arguments: 
//    stmtNo       in    The statement number (starting at 0)
//
//  Return:              vector of string values.  
//                       Empty vector if stmt does not exits. 
//
//  Contact:   N. West

  std::vector<std::string> vec;
  if ( stmtNo >= this->GetNumSqlStmts() ) return vec;

  // Locate statement
  std::string str = this->GetStmt(stmtNo);

  // Extract ...(list-of-values)... and parse it into tokens.
  std::string::size_type pos = str.find('(');
  ++pos;
  std::string::size_type n = str.find(')') - pos;
  UtilString::StringTok(vec,str.substr(pos,n),",");

  return vec;

}

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

Bool_t DbiSqlValPacket::IsEqual(const DbiSqlValPacket& that,
                                Bool_t log,
                                const Char_t* thisName,
                                const Char_t* thatName ) const {
//
//
//  Purpose:  Test for equality to another DbiSqlValPacket
//
//  Arguments: 
//    that         in    The other DbiSqlValPacket to be compared.
//    log          in    If true list differences to MSG("Dbi",kInfo)
//    thisName     in    Optional name for this packet (default: this)
//    thatName     in    Optional name for that packet (default: that)
//
//  Return:    kTRUE if identical (apart from InsertDate).
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Compare to another DbiSqlValPacket.

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

//  The InsertDate reflects the date when the data was written into the
// local database so has to be excluded from the comparison.

  if (    fSeqNo           != that.fSeqNo
       || fTableName       != that.fTableName
       || fNumStmts != that.fNumStmts ) {
    if ( log ) MSG("Dbi",Msg::kInfo) << "Conflict found:"
             << " for : " << thisName << " , " << thatName << " :-\n"
             << "  SeqNo " << fSeqNo << "," << that.fSeqNo 
             << "\n  TableName " << fTableName << "," << that.fTableName
             << "\n  Size " << fNumStmts << "," 
             << that.fNumStmts << endl;
    return kFALSE;
  }

  list<string>::const_iterator itrThisBegin = fSqlStmts.begin();
  list<string>::const_iterator itrThisEnd   = fSqlStmts.end();
  list<string>::const_iterator itrThatBegin = that.fSqlStmts.begin();
  list<string>::const_iterator itrThatEnd   = that.fSqlStmts.end();

  list<string>::const_iterator itrThis = itrThisBegin;
  list<string>::const_iterator itrThat = itrThatBegin;

  Bool_t isEqual = kTRUE;

  // Strip off InsertDate from first statement (assume its
  // the last parameter in list).

  string strThis = (*itrThis).substr(0,(*itrThis).rfind(','));
  string strThat = (*itrThat).substr(0,(*itrThat).rfind(','));
  if ( strThis != strThat ) {
  if ( ! log ) return kFALSE;
  isEqual = kFALSE;
  MSG("Dbi",Msg::kInfo) 
      << "Difference on VLD record " << ":-\n"
      << "  " << thisName << ": " << strThis  << endl
      << "  " << thatName << ": " << strThat  << endl;
  }

  // Rows can come in any order (after the first) so we have
  // to sort before comparing.  However, if we are lucky, they
  // may be in the same order or inverse order so do a quick 
  // test to see if there are no conflicts assuming these 
  // relative orderings.

  ++itrThis;
  ++itrThat;
  while ( itrThis != itrThisEnd && (*itrThis) == (*itrThat) ) {
//     cout << "Debug: trying forward compare ..." << *itrThis 
//       << "::" << *itrThat << endl;
    ++itrThis;
    ++itrThat;
  }
  if ( itrThis == itrThisEnd ) return isEqual;

  itrThis = itrThisBegin;
  itrThat = itrThatEnd;
  ++itrThis;
  --itrThat;

  while ( itrThis != itrThisEnd &&  (*itrThis) == (*itrThat) ) {
//   cout << "Debug: trying reverse compare ..." << *itrThis 
//       << "::" << *itrThat << endl;
    ++itrThis;
    --itrThat;
  }
  if ( itrThis == itrThisEnd ) return isEqual;

  // O.K., we are out of luck so set up pointers to both sets
  // and sort these.

  typedef vector<const string*>      shadow_list_t;
  typedef shadow_list_t::iterator  shadow_list_itr_t;

  shadow_list_t shadowThis;
  shadow_list_t shadowThat;

  itrThis = itrThisBegin;
  itrThat = itrThatBegin;
  ++itrThis;
  ++itrThat;

  while ( itrThis != itrThisEnd ) {
    shadowThis.push_back(&(*itrThis));
    shadowThat.push_back(&(*itrThat));
    ++itrThis;
    ++itrThat;
  }

  shadow_list_itr_t shadowThisItr    = shadowThis.begin();
  shadow_list_itr_t shadowThisItrEnd = shadowThis.end();
  shadow_list_itr_t shadowThatItr    = shadowThat.begin();
  shadow_list_itr_t shadowThatItrEnd = shadowThat.end();

  sort(shadowThisItr,shadowThisItrEnd,compStringPtrs);
  sort(shadowThatItr,shadowThatItrEnd,compStringPtrs);

  while ( shadowThisItr != shadowThisItrEnd ) {
    if ( (**shadowThisItr) != (**shadowThatItr) ){
      if ( ! log ) return kFALSE;
      isEqual = kFALSE;
      MSG("Dbi",Msg::kInfo) << "Difference on data record "
         << ":-\n"
         << "  " << thisName << ": " << **shadowThisItr  << endl
         << "  " << thatName << ": " << **shadowThatItr  << endl;
    }
   ++shadowThisItr;
   ++shadowThatItr;
  }

  return isEqual;

}

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

void DbiSqlValPacket::Print(Option_t * /* option */) const {
//
//
//  Purpose:  Print the current state.

  MSG("Dbi",Msg::kInfo) 
    << "DbiSQLValPacket:"
    << " table \"" << fTableName << "\" "
    << " SeqNo " << fSeqNo
    << " NumErrors " << fNumErrors
    << endl
    << "   CreationDate " << fCreationDate
    << endl;

  MSG("Dbi",Msg::kInfo) 
    << "   MySQL Main table creation: \"" << fSqlMySqlMetaMain << "\"" << endl;

  MSG("Dbi",Msg::kInfo) 
    << "   MySQL VLD table creation: \"" << fSqlMySqlMetaVld << "\"" << endl;

  if ( GetNumSqlStmts() > 0 ) {
    std::list<string>::const_iterator itr    = fSqlStmts.begin();
    std::list<string>::const_iterator itrEnd = fSqlStmts.end();
    for (; itr != itrEnd; ++itr) 
      MSG("Dbi",Msg::kInfo) 
        << "   SqlStmt \"" << *itr << "\"" << endl;
  }
  else {
    MSG("Dbi",Msg::kInfo) << "   No SqlStmts." << endl;
  }

}


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

void DbiSqlValPacket::Recreate(const string& tableName,
                               const VldRange& vr,
                               Int_t aggNo,
                               Dbi::Task task,             /*  Default: 0 */
                               VldTimeStamp creationDate   /*  Default: now */ 
                               ) {
//
//
//  Purpose:  Recreate and define first row (VLD - the validity record).


  this->Clear();

  DbiTableProxyRegistry& tablePR = DbiTableProxyRegistry::Instance();
  if ( ! tablePR.GetCascader().TableExists(tableName) ) {
    MAXMSG("Dbi",Msg::kError,20) << "Cannot create packet - table " << tableName
                           << " does not exist." << endl;
    fNumErrors = 1;
    return;
  }

  fTableName = tableName;


  // Create a DbiValidityRec from the supplied data.
  DbiValidityRec vrec(vr,task,aggNo,0);

  //  Create a DbiOutRowStream that can serialise this validity record
  DbiConfigSet dummy;     // For validity row any DbiTableRow will do.
  const DbiTableMetaData&  metaValid = tablePR.GetTableProxy(tableName,&dummy)
                                              .GetMetaValid();
  DbiOutRowStream buff(&metaValid);

  vrec.Store(buff,0);
  this->AddRow(buff.GetCSV());
  this->SetCreationDate(creationDate);

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

void DbiSqlValPacket::Report(const char* msg, 
                             UInt_t lineNum,
                             const string& line) {
//
//
//  Purpose:  Report and count errors.
//
//  Arguments: 
//    msg          in    Message to be reported.
//    lineNum      in    Line number.
//    line         in    Input line causing error.
//
//  Return:    
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Reset object, report and count errors.

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

//  None.

  MAXMSG("Dbi",Msg::kError,20) << msg << " on line " << lineNum
                         <<":- \n  " << line << endl;
  this->Reset();
  ++fNumErrors;

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

void DbiSqlValPacket::Reset() {
//
//
//  Purpose:  Clear out back to unfilled status except for fNumErrors.

  fSeqNo       = 0;
  fSqlMySqlMetaMain = "";
  fSqlMySqlMetaVld  = "";
  fSqlStmts.clear();
  fNumStmts    = 0;
  fTableName   = "";

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

void DbiSqlValPacket::SetCreationDate(VldTimeStamp ts) {
//
//
//  Purpose:  Set creation date.

  fCreationDate = ts;

  //  Update the validity row assuming:  "...,'creationdate',insertdate);"
  if ( this->GetNumSqlStmts() == 0 ) return;

  string& vldRow = *fSqlStmts.begin();
  string::size_type locEnd = vldRow.rfind(',');
  if ( locEnd == string::npos ) return;
  locEnd -=2;
  string::size_type locStart = vldRow.rfind(',',locEnd);
  if ( locStart == string::npos ) return;
  locStart+=2;
  vldRow.replace(locStart,locEnd-locStart+1,ts.AsString("s"));

}

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

void DbiSqlValPacket::SetMetaData() const {
//
//
//  Purpose:  Set up meta-data as SQL table creation statements.

  DbiTableProxyRegistry& tbprxreg = DbiTableProxyRegistry::Instance();

  //  Locate the table in the cascade.
  DbiCascader& cas = tbprxreg.GetCascader();
  Int_t dbNo = cas.GetTableDbNo(this->GetTableName());
  if ( dbNo < 0 ) return;

  //  Any table proxy will do to get the meta-data so use the one for a 
  //  DbiConfigSet;
  DbiConfigSet dummy;
  const DbiTableMetaData & metaVld =  tbprxreg.GetTableProxy(this->GetTableName(),&dummy)
                                     .GetMetaValid();
  const DbiTableMetaData & metaMain = tbprxreg.GetTableProxy(this->GetTableName(),&dummy)
                                     .GetMetaData();
  fSqlMySqlMetaVld   = metaVld.Sql();
  fSqlMySqlMetaMain  = metaMain.Sql();

}

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

void DbiSqlValPacket::SetSeqNo(UInt_t seqno) {
//
//
//  Purpose:  Set Sequence number.

  fSeqNo = seqno;

  //  Update all rows 
  if ( this->GetNumSqlStmts() == 0 ) return;

  ostringstream tmp;
  tmp << seqno;
  const string seqnoStr = tmp.str();

  std::list<string>::iterator itr    = fSqlStmts.begin();
  std::list<string>::iterator itrEnd = fSqlStmts.end();
  for (; itr != itrEnd; ++itr) SetSeqNoOnRow(*itr,seqnoStr);


}

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

void DbiSqlValPacket::SetSeqNoOnRow(string& row,const string& seqno) {
//
//
//  Purpose:  Set Sequence number on supplied row

//  Update row assuming:  "...(seqno, ...."
  
  string::size_type locStart = row.find('(');
  if ( locStart == string::npos ) return;
  ++locStart;
  string::size_type locEnd = row.find(',',locStart);
  if ( locEnd == string::npos ) return;
  row.replace(locStart,locEnd-locStart,seqno);

}

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

Bool_t DbiSqlValPacket::Store(UInt_t dbNo, Bool_t replace) const {
//
//
//  Purpose:  Output validity packet to specified database.
//
//  Arguments: 
//    dbNo         in    Number of database in the cascade.
//    replace      in    If true replace existing SeqNo (default: kFALSE).
//
//  Return:    kTRUE if successfully stored.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Output validity packet to specified database modifying
//    InsertDate to be current date.

  if ( ! CanBeStored() ) return kFALSE; 

  //Just use any old table row object just to get a DbiDBProxy.
  DbiConfigSet pet;
  DbiTableProxy& tp =  DbiTableProxyRegistry::Instance()
                      .GetTableProxy(this->GetTableName(),&pet);
  if ( replace ) {
    const DbiDBProxy & proxy = tp.GetDBProxy();
    if ( ! proxy.RemoveSeqNo(this->GetSeqNo(),dbNo) ) return kFALSE;
  }

  // Locate required DbiStatement.
  auto_ptr<DbiStatement> stmtDb(DbiTableProxyRegistry::Instance()
                               .GetCascader()
                               .CreateStatement(dbNo));
  if ( ! stmtDb.get() ) {
    MSG("Dbi",Msg::kWarning) 
      << "Attempting to write to non-existant cascade entry " << dbNo
      << endl;
    return kFALSE;
  }

  // Decide if ROW_COUNTER column has to be removed i.e. Master
  // DB exported with ROW_COUNTER but Slave still has old table form
  // without ROW_COUNTER.

  bool removeRowCounter =    fgLastMetaHadRowCounter 
                          && ! tp.GetMetaData().HasRowCounter();

  // Loop processing all SQL statements
  Bool_t first = kTRUE;
  int combineInserts = 0;
  // Cannot group ORACLE inserts.
  int maxInserts = stmtDb->GetDBType() == Dbi::kOracle ? 1 : 20;  
  string sqlInserts;

  for (list<string>::const_iterator itr = fSqlStmts.begin();
       itr != fSqlStmts.end();
       ++itr) {
    if ( first ) {
//    On first statement replace InsertDate by current datetime.
      string sql = *itr;
      list<string>::size_type locDate = sql.rfind(",\'");
      if ( locDate !=  string::npos ) {
        VldTimeStamp now;
        sql.replace(locDate+2,19,Dbi::MakeDateTimeString(now));
      }
      stmtDb->ExecuteUpdate(sql.c_str());
      if ( stmtDb->PrintExceptions() ) return kFALSE;
      first = kFALSE;
      continue;
    }

    string sql = *itr;

//  On other statements remove the second (ROW_COUNTER) column if required.
    if (removeRowCounter) {
      list<string>::size_type locStart  = sql.find(',');
      ++locStart;
      list<string>::size_type locEnd    = sql.find(',',locStart);
      if ( locEnd !=  string::npos ) {
        list<string>::size_type numChars = locEnd - locStart + 1;
        sql.erase(locStart,numChars);
      }
    }

//  Reduce database I/O by combining groups of insert commands.

    string::size_type insertIndex = sql.find("VALUES (");
    if ( insertIndex == string::npos) {
      MAXMSG("Dbi",Msg::kError,20) << "Unexpected SQL : " << sql
                             << "\n  should be of form INSERT INTO ... VALUES (...);" << endl;
      return kFALSE;
    }
    ++combineInserts;
    if ( combineInserts == 1 ) sqlInserts = sql;
    else {
      sqlInserts[sqlInserts.size()-1] = ',';
      sqlInserts += sql.substr(insertIndex+7);
    }
    if ( combineInserts >= maxInserts ) {
      stmtDb->ExecuteUpdate(sqlInserts.c_str());
      if ( stmtDb->PrintExceptions() ) return kFALSE;
      combineInserts = 0;
    }
  }

// Deal with last group of inserts.
  if ( combineInserts ) {
    stmtDb->ExecuteUpdate(sqlInserts.c_str());
    combineInserts = 0;
    if ( stmtDb->PrintExceptions() ) return kFALSE;
  }

  return kTRUE;

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

Bool_t DbiSqlValPacket::Write(std::ofstream& ios,
                              Bool_t addMetadata) const {
//
//
//  Purpose:  Export to an iostream.
//
//  Arguments: 
//     ios          in/out    Output stream.
//     addMetadata  in        if kTRUE, output SQL metadata (in form
//                            of a create table SQL)
//
//  Return:    kTRUE is I/O successful.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Export to an iostream.

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

//  None.

  if ( ! CanBeStored() ) return kFALSE;
  if ( addMetadata ) {
    if ( fSqlMySqlMetaMain.size() == 0 ) this->SetMetaData();
    if ( fSqlMySqlMetaMain.size() == 0 ) {    
      MSG("Dbi",Msg::kWarning) 
      << "Cannot write metadata; no associated DbiTableProxy "
      << endl;
    }
    else {


      ios << ">>>>>" << GetTableName() << " Metadata [MySQL]" << endl;
      ios << fSqlMySqlMetaVld   << endl;
      ios << fSqlMySqlMetaMain  << endl;
      ios << "<<<<<" << GetTableName() << " Metadata" << endl;
    }
  }

  ios << ">>>>>" << GetTableName() << "  " << GetSeqNo() << endl;

  for ( list<string>::const_iterator itr = fSqlStmts.begin();
        itr != fSqlStmts.end();
        ++itr) ios << (*itr) << endl;

  ios << "<<<<<" << GetTableName() << "  " << GetSeqNo() << endl;    

  return kTRUE;

}

---