DbmModule.cxx

Go to the documentation of this file.

// Dbm                                                                //
//                                                                    //
// package: Dbm (Database Maintenance).                               //
//                                                                    //
// Driver module.                                                     //
//                                                                    //
// N. West  08/2001                                                   //

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

#include "Conventions/Detector.h"
#include "Conventions/SimFlag.h"

#include "DatabaseInterface/Dbi.h"
#include "DatabaseInterface/DbiExceptionLog.h"
#include "DatabaseInterface/DbiDBProxy.h"
#include "DatabaseInterface/DbiCascader.h"
#include "DatabaseInterface/DbiConfigSet.h"
#include "DatabaseInterface/DbiResultKey.h"
#include "DatabaseInterface/DbiResultPtr.h"
#include "DatabaseInterface/DbiSqlValPacket.h"
#include "DatabaseInterface/DbiStatement.h"
#include "DatabaseInterface/DbiTableProxyRegistry.h"
#include "DatabaseInterface/DbiTimerManager.h"
#include "DatabaseInterface/DbiValidityRec.h"
#include "DatabaseInterface/DbiValRecSet.h"
#include "DatabaseInterface/DbiTableProxy.h"
#include "DatabaseInterface/DbiTableProxyRegistry.h"
#include "DatabaseInterface/DbiWriter.h"

#include "DatabaseMaintenance/DbmCmdOptions.h"
#include "DatabaseMaintenance/DbmLogFile.h"
#include "DatabaseMaintenance/DbmModule.h"
#include "DatabaseMaintenance/DbmNameFilter.h"

#include "LeakChecker/LeaLeakChecker.h"
#include "JobControl/JobCModuleRegistry.h" 
#include "JobControl/JobCommand.h" 

#include "MessageService/MsgService.h"     

#include "TSQLStatement.h"

#include "Util/MsgUtil.h"
#include "Validity/VldContext.h"
#include "Validity/VldTimeStamp.h"


using namespace std;

ClassImp(DbmModule)

CVSID("$Id: DbmModule.cxx,v 1.63 2007/11/11 07:41:21 rhatcher Exp $");

JOBMODULE(DbmModule,"DbmModule","Database Maintenance");

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

DbmModule::DbmModule() :
fCascader(DbiTableProxyRegistry::Instance().GetCascader()),
fValRecSet(0)
{
//
//
//  Purpose:  Default constructor
//
//  Arguments:  None.
//
//  Return:    n/a
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o  Create a DbmModule.


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

//  None.

  LEA_CTOR    //Leak Checker

    //  MsgService::Instance()->GetStream("Dbi")->SetLogLevel(Msg::kVerbose);
    //  MsgService::Instance()->GetStream("Dbm")->SetLogLevel(Msg::kDebug);
  MSG("Dbm", Msg::kVerbose) << "Creating DbmModule" << endl;
  
}

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

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


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

//  None.

  LEA_DTOR    //Leak Checker

  MSG("Dbm", Msg::kVerbose) << "Destroying DbmModule" << endl;

  ClearCache();

}

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

Bool_t DbmModule::ApplySQLFixup(UInt_t dbNo,
                                ifstream& in, 
                                DbmLogFile& logFile,
                                Bool_t enabled) {
//
//
//  Purpose:  Apply any SQL fixups at the start of update file in and log
//
//               SQL fixup lines start: SQL:
//
//            Return true if no errors detected.

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

  // Use peek to look one character ahead without disturbing input;
  // we must not read the start of the first update record.

  while ( in.peek() == 'S' ) {
    string line;
    getline(in,line);
    if ( line.substr(0,4) != "SQL:" ) {
      MSG("Dbm",Msg::kError) 
        << "SQL Fixup line does not start: SQL: " << line
        << endl;
     return kFALSE;
    }

    //Remove anything after last ';'
    string::size_type locSemiColon = line.rfind(';');
    if ( locSemiColon == string::npos )  {
      MSG("Dbm",Msg::kError) 
        << "SQL Fixup line does not end with semicolon: " << line
        << endl;
     return kFALSE;
    }
    line.erase(locSemiColon+1);

    if ( enabled ) {
      //some hackery to make sure data gets deleted before VLD entry
      std::string::size_type p;
      if((line.find("delete from ")!=std::string::npos) &&
         ((p=line.find("VLD"))!=std::string::npos)){
        std::string noVLD=line.substr(0,p);
        noVLD+=line.substr(p+3);
        stmtDb->ExecuteUpdate(noVLD.substr(4).c_str());
        logFile.LogMsg(line,kFALSE);
        if ( stmtDb->PrintExceptions() ) logFile.LogMsg("   - above command failed.",kFALSE);
      }
      stmtDb->ExecuteUpdate(line.substr(4).c_str());
      logFile.LogMsg(line,kFALSE);
      if ( stmtDb->PrintExceptions() ) logFile.LogMsg("   - above command failed.",kFALSE);
    }
    else {
      string msg("SQL fixup disabled. Ignoring :");
      msg += line;
      logFile.LogMsg(msg,kFALSE);
    }

  }
  return kTRUE;

}

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

Bool_t DbmModule::BuildValRecSet(const string& tableName,UInt_t dbNo) {
//
//
//  Purpose:  If necessary, rebuild DbiValidityRec for selected table.
//
//  Arguments: 
//    tableName    in    Name of selected table.
//    dbNo         in    Cascade number of Database to be used.
//
//  Return:    kTRUE if table successfully built (or already exists)
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o If fValRecSet does not point at DbiValidityRec of requested
//    table then rebuild it.

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

//  fValRecSet is just a cache to reduce the number of builds
//  of a DbiValRecSet for a given table.  For example the Export
//  method requires it but calls the GlobaliseSeqNo method which
//  also requires it.

  if (    fValRecSet 
       && (    fValRecSet->GetTableName() != (tableName+"VLD") 
            || fValRecSet->GetDbNo()      != dbNo ) ) {
    MSG("Dbm", Msg::kDebug) << "(Dropping Table "
                           << fValRecSet->GetTableName() << ")" << endl;
    ClearCache();
  }

  if ( ! fValRecSet ) {
    MSG("Dbm", Msg::kDebug) << "Reading Table " << tableName
                           << "..." << endl;
    
    fValRecSet = new DbiValRecSet(tableName,dbNo);
  }

  int numRows = fValRecSet->GetNumRows();
  if ( numRows ) {
    MSG("Dbm", Msg::kInfo) << "Table " << tableName
                           << " contains " << numRows
                           << " sequence numbers" << endl;
  }
  else {
    MSG("Dbm", Msg::kDebug) << "Table " << tableName
                           << " missing or empty!" << endl;
  }

  return numRows > 0;

}

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

Bool_t DbmModule::CheckMemory(JobCommand *cmd) {
//
//
//  Purpose:  Check memory
//
//    cmd          in    Command parameters. Optional, default = 0;       
//  Arguments: None.
//
//  Return:    kTRUE.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Check memory.

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

//  None.


// Check command options, if supplied.

  if ( cmd ) {
    DbmCmdOptions opts("",cmd);
    if ( ! opts.IsValid() ) return kFALSE;
  }

// Clear out local cache to make any leak easier to spot.

  ClearCache();

  LeaLeakChecker* lea = LeaLeakChecker::Instance();
  MSG("Dbm",Msg::kInfo) << lea << endl;
  DbiTableProxyRegistry::Instance().ShowStatistics();

  return kTRUE;

}

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

void DbmModule::ClearCache() {
//
//
//  Purpose:  Clear out cached  DbiValidityRec.
//
//  Arguments: None.

//
//  Return:    n/a
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Clear out cached  DbiValidityRec.
//
//  Program Notes:-
//  =============

//  None.

  delete fValRecSet;
  fValRecSet = 0;

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

Bool_t DbmModule::ExportTable(const string& tableName,
                              ofstream* out,
                              DbmLogFile& logFile,
                              list<int>seqnos) {
//
//
//  Purpose:  Export the contents of the specified table to an ostream.
//
//  Arguments: 
//    tableName    in    The table to operate on.
//    out          in    The output stream.
//    logFile      in    Log file to record I/O,
//    seqnos       in    List of selected sequence numbers.
//                       If empty, export all.
//
//  Return:    kTRUE if output successful.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Ensure that all exported entries have global SeqNos.
//
//  o Loop over all entries in the auxillary validity range table and
//    export each to a file.
//  
//  o For any entry whose insertion date is the far future
//    (year >= 2030), set it back a few minutes before the current date.
//
//  o For first entry add SQl to create table.

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

//  The DbiConfigSet table row is used as a generic row to export all
//  types of tables.

//  The reseting of far future insertion dates allows people to add
//  data into the database with dummy insertion date and have it
//  exported as if the insertion date were correct.

// Check that all have global SeqNos (will also ensure that fValRecSet
// will contain all DbiValidityRecs for table).


  MSG("Dbm", Msg::kInfo) << "\nExporting table " 
                         << tableName << "..." << endl;

  if ( ! BuildValRecSet(tableName,0) ) return kFALSE;

  int numVRecs    = fValRecSet->GetNumRows();
  int numVRecsOut = 0;
  Int_t numModified = 0;

  if ( numVRecs == 0 ) {
    MSG("Dbm", Msg::kInfo) << "Nothing to do: table " 
                           << tableName << " is empty!" << endl;
    return kTRUE; 
  }

  VldTimeStamp farFuture(2030,1,1,0,0,0);
// Now is actually 5 mins ago otherwise next Export will
// pick it up again.
  VldTimeStamp now(VldTimeStamp().GetSec() - 5*60, 0);
  
  const DbiDBProxy& proxy = GetDBProxy(tableName);

  Int_t numDataRows = 0;
  Bool_t writeMeta  = kTRUE;

  for ( int jvrec = 0; jvrec < numVRecs; ++jvrec ) {

    // Create an inner scope to limit the lifetime of packet
    {

      // Skip if unwanted.
      const DbiValidityRec* vrec = fValRecSet->GetTableRow(jvrec);
      int seqNo   = vrec->GetSeqNo();
      bool isWanted = true;
      if ( seqnos.size() > 0 ) {
        isWanted = false;
        int seqNoMin = 999999999;
        list<int>::iterator  itr(seqnos.begin()), itrEnd(seqnos.end());
        for ( ; itr != itrEnd; ++itr ) {
          int seqNoMax = *itr;
          if ( seqNoMax < 0 ) {
            seqNoMax = - seqNoMax;
            if ( seqNo >= seqNoMin && seqNo <= seqNoMax ) isWanted = true;
          }
          else {
            if ( seqNo == seqNoMax ) isWanted = true; 
          }
          seqNoMin = seqNoMax;
        }
      }
      if ( ! isWanted ) continue;
      DbiSqlValPacket packet(*vrec);

      // Deal with far future insert dates.
      if ( vrec->GetInsertDate() > farFuture ) { 
          vrec->GetTableProxy()
                 ->GetDBProxy()
                .ReplaceInsertDate(now,seqNo,vrec->GetDbNo());
      }

      // Renumber local SEQNOs.
      if ( seqNo <= Dbi::kMAXLOCALSEQNO ) {
        Int_t globalSeqNo = fCascader.AllocateSeqNo(tableName,1);
        if ( globalSeqNo < Dbi::kMAXLOCALSEQNO ) {
          MSG("Dbi",Msg::kError) << "Failed (cascader returned " <<globalSeqNo
                                 << ") to get replacement for " 
                                 << seqNo << endl;
          continue;
        }
  
        MSG("Dbi",Msg::kInfo) << "    Renumbering " << seqNo 
                              <<  " -> " << globalSeqNo << endl;
        if ( ! proxy.ReplaceSeqNo(seqNo,globalSeqNo,0) ) continue;
        else ++numModified;
        seqNo = globalSeqNo;
        packet.SetSeqNo(seqNo);
      }

      Int_t  numRows = packet.GetNumSqlStmts()-1;

      MSG("Dbm", Msg::kInfo)
        << "Exporting sequence number " << seqNo
        << " with " << numRows << " rows of data" << endl;

      packet.Write(*out,writeMeta);
      writeMeta = kFALSE;
      logFile.LogRec(vrec);
      numDataRows += numRows;
      ++numVRecsOut;

    }
  
//  Packet will have been destroyed so can now purge caches, 
//  the data won't be needed again.
    DbiTableProxyRegistry::Instance().PurgeCaches();
  }

  // If any sequence number has been modified flush the cache.
  if ( numModified ) ClearCache();

  MSG("Dbm", Msg::kInfo)
    << "A total of " << numVRecsOut << " validity entries and "
    << numDataRows << " data rows written to file" << endl;
  // Flush out last record.
  logFile.LogRec();


  return kTRUE;

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

Bool_t DbmModule::ExportTables(JobCommand *cmd) {
//
//
//  Purpose:  Export tables..
//
//  Arguments: 
//    cmd          in    Command parameters.          
//
//  Return:    kTRUE if output successful.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Process command parameters.
//
//  o Loop over all selected tables exporting them.

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

// None.

// Process all options.

  MSG("Dbm",Msg::kInfo) << "\n\nProcessing EXPORT command...\n" << endl;

  DbmCmdOptions opts("--File:--LogFile:--SeqNo:--Since:--Table", cmd);
  if ( ! opts.IsValid() ) return kFALSE;


//  Deal with --LogFile but always create a logfile even
//  if its dummy.  That way it can be put on the stack.

  string logName;
  if ( opts.TestOpt("--LogFile") ) {
    logName = opts.GetOptString("--LogFile");
  }
  DbmLogFile logFile(logName);

//  Deal with --SeqNo.

  list<int> seqnos;
  if ( opts.TestOpt("--SeqNo" ) ) opts.GetOptIntList("--SeqNo",seqnos);

//  Deal with --Since.
  string insertDateCondition;
  if ( opts.TestOpt("--Since" ) ) {
    string opt = opts.GetOptString("--Since");
//  Have to add leading and trailing double quotes for SQL
    insertDateCondition =   insertDateCondition + "InsertDate >'" 
                            + opt + "'";
    MSG("Dbm",Msg::kInfo) << "Applying condition: " 
                          << insertDateCondition  << endl;
  }

// Deal with --File and --Table.
  if ( ! opts.TestOpt("--Table" ) ) {
    MSG("Dbm",Msg::kWarning) << "Table names  not specified." << endl;
    return kFALSE;
  }
  if ( ! opts.TestOpt("--File" ) ) {
    MSG("Dbm",Msg::kWarning) << "File name  not specified." << endl;
    return kFALSE;
  }
 
  DbmNameFilter nameFilter(opts.GetOptString("--Table"));
  MSG("Dbm",Msg::kInfo) << " Export table name mask set to: " 
                        << opts.GetOptString("--Table") <<  endl;
  string fileName = opts.GetOptString("--File");


//  Open export file.
  ofstream exportFile(fileName.c_str(),ios_base::app);
  if ( ! exportFile ) {
    MSG("Dbm", Msg::kInfo) << "Unable to open " << fileName << endl;
    return kFALSE; 
  }


  //Collect list of table names to be processed before proceesing
  // because TSQLResultSet only valid until next query.
    list<string> tableNames;
    this->ListTables(nameFilter,0,tableNames);

  // If a start date is to be applied clear out any cached 
  // DbiValidityRecs before applying.
  if ( insertDateCondition != "" ) {
    ClearCache();
    DbiTableProxyRegistry::Instance()
                           .SetSqlCondition(insertDateCondition);
  }

  // Now process all tables.

  logFile.LogCmd("export",opts);

  for ( list<string>::const_iterator itr = tableNames.begin();
        itr != tableNames.end();
        ++itr ) {
    DbiTableProxyRegistry::Instance()
                           .SetSqlCondition(insertDateCondition);
    ExportTable(*itr,&exportFile,logFile,seqnos);
    DbiTableProxyRegistry::Instance().SetSqlCondition("");
  }

  // Undo any SQL conditions.
  if ( insertDateCondition != "" ) {
    ClearCache();
    DbiTableProxyRegistry::Instance().SetSqlCondition();
  }

//  Close export file (even though it will as it goes out of scope).
  exportFile.close();

  return kTRUE;

}

//.....................................................................
const DbiDBProxy& DbmModule::GetDBProxy(const string& tableName) const {
//
//
//  Purpose:  Return a DBProxy for specified table.
//
//  Arguments: 
//    tableName    in   Name of table.
//
//  Return:    
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Return a DBProxy for specified table.

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

//  For all tables Dbm uses DbiConfigSet as a generic table row.

  DbiConfigSet pet;
  return  DbiTableProxyRegistry::Instance()
            .GetTableProxy(tableName,&pet)
               .GetDBProxy(); 

}

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

Bool_t DbmModule::GlobaliseSeqNo(JobCommand *cmd) {
//
//
//  Purpose:  Convert to global any local sequence numbers in supplied
//            table name.
//
//  Arguments: 
//    cmd          in    Command parameters.          
//
//  Return:    kTRUE if all sequence numbers are now global.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Ensure that there is an authorising database is the cascade.
//
//  o Load all entries in the auxillary validity range table and
//    look for local sequence numbers ( <= Dbi::kMAXLOCALSEQNO)
//    and for each convert to global in both this and the main table.
//

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

// None.

// Process all options.

  string tableName;

  DbmCmdOptions opts("--Table", cmd);
  if ( ! opts.IsValid() ) return kFALSE;

  if ( ! opts.TestOpt("--Table" ) ) {
    MSG("Dbm",Msg::kWarning) << "Table name  not specified." << endl;
    return kFALSE;
  }

  tableName = opts.GetOptString("--Table");

  return GlobaliseSeqNo(tableName);

}

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

Bool_t DbmModule::GlobaliseSeqNo(const string& tableName) {
//
//
//  Purpose:  Convert to global any local sequence numbers in supplied
//            table name.
//
//  Arguments: 
//    tableName    in    The table to be globalised.
//
//  Return:    kTRUE if all sequence numbers are now global.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Ensure that there is an authorising database is the cascade.
//
//  o Load all entries in the auxillary validity range table and
//    look for local sequence numbers ( <= Dbi::kMAXLOCALSEQNO)
//    and for each convert to global in both this and the main table.
//

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

//  As a by product this sets up fValRecSet.

  Bool_t ok         = kTRUE;
  Int_t numModified = 0;

  // Collect up all DbiValityRecs.
  if ( ! BuildValRecSet(tableName,0) ) return kFALSE;

  // Make sure that there is an authorising database.
  if ( fCascader.GetAuthorisingDbNo() < 0 ) {
    MSG("Dbi",Msg::kError) << "No authorising DB in cascade" << endl;
    return kFALSE;
  }

  // Loop over all DbiValidityRecs checking for and converting local SeqNos

  const DbiDBProxy& proxy = GetDBProxy(tableName);

  MSG("Dbm", Msg::kInfo) << "Scanning for local sequence numbers..." << endl;
  int numRows = fValRecSet->GetNumRows();

  for ( int irow = 0; irow < numRows; ++ irow ) {
    const DbiValidityRec* vrec = fValRecSet->GetTableRow(irow);
    int seqNo = vrec->GetSeqNo();
    if ( seqNo <= Dbi::kMAXLOCALSEQNO ) {
      Int_t globalSeqNo = fCascader.AllocateSeqNo(tableName,1);
      if ( globalSeqNo < Dbi::kMAXLOCALSEQNO ) {
        MSG("Dbi",Msg::kError) << "Failed to get replacement for " 
                               << seqNo << endl;
        ok = kFALSE;
        break;
      }

      MSG("Dbi",Msg::kInfo) << "    Renumbering " << seqNo 
                            <<  " -> " << globalSeqNo << endl;
      if ( ! proxy.ReplaceSeqNo(seqNo,globalSeqNo,0) ) ok = kFALSE;
      else ++numModified;
      
    }
  }

  MSG("Dbi",Msg::kInfo) << "A total of " << numRows
                          << " were scanned" << endl;

  // If any sequence number has been modified flush the cache.
  if ( numModified ) {
    MSG("Dbi",Msg::kInfo) << "A total of " << numModified
                          << " were renumbered" << endl;
    ClearCache();
    BuildValRecSet(tableName,0);
  }
  else MSG("Dbi",Msg::kInfo) << "No local SeqNos found" << endl;
  return ok;

}

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

void DbmModule::HandleCommand(JobCommand *cmd) {
//
//
//  Purpose:  Command Processor
//
//  Arguments: 
//    cmd          in    JobCommand holding command to be processed.
//
//  Return:    n/a
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Execute supplied database maintenance command.

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

//  None.

// Clear cache in case previous command modified
// the table is was caching.

  this->ClearCache();

  const char* c = cmd->PopCmd();

  if (c) {
    TString sc(c);
    sc.ToLower();
    if      (sc == "checkmemory" )    CheckMemory(cmd);
    else if (sc == "export" )         ExportTables(cmd);
    else if (sc == "globaliseseqno" ) GlobaliseSeqNo(cmd);
    else if (sc == "import" )         Import(cmd);
    else if (sc == "logentry" )       LogEntry(cmd);
    else if (sc == "help" )           Help();
    else if (sc == "query" )          Query(cmd);
    else if (sc == "setloglevel" )    SetLogLevel(cmd);
    else if (sc == "settimer" )       SetTimer(cmd);

    else {
      MSG("Dbm",Msg::kWarning) << "Unknown command " << sc <<endl
           << "Type Help for valid commands." << endl;

    }
  }

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

void DbmModule::Help() {
//
//
//  Purpose:  Print help.
//
//  Arguments: None.
//
//  Return:  none.  
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o  Print help.

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

//  None.

  string helpFileName 
     = fFileLocater.Find("DatabaseMaintenance/doc/Help.txt");
    ifstream helpFile(helpFileName.c_str());

  if ( ! helpFile.is_open() ) {
    MSG("Dbm",Msg::kInfo) << "Sorry, I cannot help you!" << endl;
  }
  else {
    string line;
    while ( ! helpFile.eof() ) {
      getline(helpFile,line);
      MSG("Dbm",Msg::kInfo) << line << endl;
    }
  }


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

Bool_t DbmModule::Import(JobCommand *cmd) {
//
//
//  Purpose:  Import the contents of the specified file into database.
//
//  Arguments: 
//    cmd          in    Command parameters.          
//
//  Return:    kTRUE if import successful.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Process input file, skipping any local or duplicate SeqNos 
//    and ignoring excluded tables.
//
//  o If table does not exist, get DbiSqlvalPacket to create them
//    usin the create table SQL.


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


  MSG("Dbm",Msg::kInfo) << "\n\nProcessing IMPORT command...\n" << endl;

// Process all options.
  DbmCmdOptions opts("--Batch:"
                     "--DiscardConflicts:"
                     "--DatabaseNumber:"
                     "--EnableSQL:"
                     "--File:"
                     "--LogFile:"
                     "--Table:"
                     "--Test",
                     cmd);
  if ( ! opts.IsValid() ) return kFALSE;

//  Deal with --DatabaseNumber
  UInt_t dbNo = 0;
  if ( opts.TestOpt("--DatabaseNumber" ) ) {
    dbNo = opts.GetOptInt("--DatabaseNumber");
    MSG("Dbm",Msg::kInfo) << " using database number " << dbNo
                          << "  " << opts.GetOptString("--DatabaseNumber")
                        << " for import" <<  endl;
  }

//  Deal with --Batch
   Bool_t batch = opts.TestOpt("--Batch");
   if ( batch ) MSG("Dbm",Msg::kInfo) 
       << " Batch mode: Errors will force exit at end of command." <<  endl;

//  Deal with --DiscardConflicts
   Bool_t discardConflicts = opts.TestOpt("--DiscardConflicts");
   if ( discardConflicts ) MSG("Dbm",Msg::kInfo) 
       << " Conflicts will be discarded." <<  endl;
   else MSG("Dbm",Msg::kInfo) 
       << " Conflicts will be imported." <<  endl;

//  Deal with --EnableSQL
   Bool_t enableSQL = opts.TestOpt("--EnableSQL");
   if ( enableSQL ) MSG("Dbm",Msg::kInfo) 
       << " SQL fixups are enabled." <<  endl;
   else MSG("Dbm",Msg::kInfo) 
       << " SQL fixups are disabled." <<  endl;

//  Deal with --File
  if ( ! opts.TestOpt("--File" ) ) {
    MSG("Dbm",Msg::kWarning) << "File name  not specified." << endl;
    if ( batch ) this->Quit();
    return kFALSE;
  }
  string fileName = opts.GetOptString("--File");

//  Deal with --LogFile but always create a logfile even
//  if its dummy.  That way it can be put on the stack.
  string logName;
  if ( opts.TestOpt("--LogFile") ) {
    logName = opts.GetOptString("--LogFile");
  }
  DbmLogFile logFile(logName,dbNo);


//  Deal with optional --Table
  string tableMask = "*";
  if ( opts.TestOpt("--Table") ) tableMask = opts.GetOptString("--Table");
  DbmNameFilter nameFilter(tableMask);
  MSG("Dbm",Msg::kInfo) << " Import table name mask set to: " << tableMask <<  endl;

//  Deal with --Test
  Bool_t test = opts.TestOpt("--Test");
  if ( test ) MSG("Dbm",Msg::kInfo) << " Testing only, database will "
                                     << " not be updated." <<  endl;  

//  Open import file.
  ifstream import(fileName.c_str());
  if ( ! import ) {
    MSG("Dbm", Msg::kError) << "Unable to open " << fileName << endl;
    if ( batch ) this->Quit();
    return kFALSE; 
  }

  MSG("Dbm", Msg::kInfo) << "\n\nProcessing file " << fileName 
                         << " ..." << endl;

  Int_t numNewTables = 0;
  Int_t numImported  = 0;
  Int_t numImpConfl  = 0;
  Int_t numImpUpdate = 0;
  Int_t numDuplicate = 0;
  Int_t numOutOfDate = 0;
  Int_t numConflict  = 0;
  Int_t numLocal     = 0;
  Int_t numFailed    = 0;
  Int_t numTest      = 0;

  string currentTableName;

  logFile.LogCmd("import",opts);

//  Apply any SQL fixups before starting the import loop.

  if ( ! this->ApplySQLFixup(dbNo, import, logFile, enableSQL) ) {
    MSG("Dbm", Msg::kError) << "Fixup failed " << endl;
    if ( batch ) this->Quit();
    return kFALSE; 
  }

// Loop repeatedly filling a DbiSqlValPacket and storing it.

  DbiSqlValPacket packetFile(import);

  Bool_t skipTable = kFALSE;

  for (;  packetFile.CanBeStored(); packetFile.Fill(import) ) {

    //  Check for table change.
    if ( currentTableName != packetFile.GetTableName() ) {
      currentTableName = packetFile.GetTableName();

      // Skip table if not in name filter.
      skipTable =  ! nameFilter.Test(currentTableName);
      if ( skipTable ) {
        if ( ! batch ) MSG("Dbm", Msg::kInfo) << "\nSkipping table " 
                             << currentTableName << "...\n" << endl;
        continue;
      }
      if ( ! batch ) MSG("Dbm", Msg::kInfo) << "\nUpdating table " 
                             << currentTableName << "...\n" << endl;

      // If it doesn't exist, try to create it.
      auto_ptr<DbiStatement> stmtDb(fCascader.CreateStatement(dbNo));
      string sql = "select * from  ";
      sql += currentTableName + " where 1=0;";
      TSQLStatement* res = stmtDb->ExecuteQuery(sql.c_str());

      if ( ! res ) {
        if ( packetFile.CreateTable(dbNo) ) {
          MSG("Dbm", Msg::kInfo) << "   Created table: "
                                 << currentTableName << endl;
          ++numNewTables;
        }
        else {
          MSG("Dbm", Msg::kError) << "   failed to create table: "
                                 << currentTableName << endl;
        }
      }
    }

    // Ignore if table to be skipped.
    if ( skipTable ) continue;

    Int_t seqNo = packetFile.GetSeqNo();

    // Decide if record is to be imported.
    bool tryImport = false;
    bool replace   = false;

    // Check for duplicates.

    DbiValRecSet vrecSet(currentTableName,dbNo,seqNo);
    const DbiValidityRec* vrec = vrecSet.GetTableRowBySeqNo(seqNo);

    if ( vrec ) {
//       DbiSqlValPacket packetDb(*vrec);
//       DbiSqlValPacket::CompResult_t comp
//      = packetDb.Compare(packetFile,kTRUE,"Database","Import  ");

      DbiSqlValPacket::CompResult_t comp = DbiSqlValPacket::kIdentical;
      if ( packetFile.GetCreationDate() > vrec->GetCreationDate() 
         )  comp = DbiSqlValPacket::kUpdate;
      if ( packetFile.GetCreationDate() < vrec->GetCreationDate() 
         )  comp = DbiSqlValPacket::kOutOfDate;

      switch ( comp ) {

      case DbiSqlValPacket::kIdentical :
        if ( ! batch ) MSG("Dbm", Msg::kInfo) << "   " << seqNo 
                                      << " skipped (duplicate)" << endl;
        ++numDuplicate;
        break;

      case DbiSqlValPacket::kUpdate :
        ++numImpUpdate;
        tryImport = true;
        replace   = true;
        break;

      case DbiSqlValPacket::kOutOfDate :
        if ( ! batch ) MSG("Dbm", Msg::kInfo) << "   " << seqNo 
                                    << " skipped (out of date)" << endl;
        ++numOutOfDate;
        break;

      // Everything else is a conflict!
      default :
        if ( discardConflicts ) {
          MSG("Dbm", Msg::kError) << "   " << seqNo 
                                      << " skipped (conflict!)" << endl;
          ++numConflict;
        }
        else {
          MSG("Dbm", Msg::kError) << "   " << seqNo 
                                      << " accepting conflict!" << endl;
          ++numImpConfl;
          tryImport = true;
          replace   = true;
        }
      }
    }
    else {
      if ( Dbi::NotGlobalSeqNo(seqNo) )  {
        ++numLocal;
        if ( ! batch ) MSG("Dbm", Msg::kInfo) << "   " << seqNo 
                                   << " skipped (local sSeqNo)" << endl;
      }

      else {
        tryImport = true;
      }
    }

//  Purge caches, used when getting DbiSqlValPackets from database;
//  the data won't be needed again.
    if ( vrec )  DbiTableProxyRegistry::Instance().PurgeCaches();

    // Try to import unless already rejected.
    if ( tryImport ) {

      // Skip importing if testing.
      if ( test ) {
        ++numTest;
        if ( ! batch ) MSG("Dbm", Msg::kInfo) << "   " << seqNo 
                                              << " tested" << endl;
      }
      else {

        if ( packetFile.Store(dbNo,replace) )  {
          ++numImported;
          if ( ! batch ) MSG("Dbm", Msg::kInfo) << "   " << seqNo 
                                                << " imported" << endl;
          logFile.LogRec(currentTableName,seqNo);
        }
        else {
        ++numFailed;
         MSG("Dbm", Msg::kError) << "   " << seqNo
                                          << " failed" << endl;
         // Attempt to delete failed SEQNO;
         DbiConfigSet pet;
         DbiTableProxyRegistry::Instance()
           .GetTableProxy(currentTableName,&pet)
             .GetDBProxy().RemoveSeqNo(seqNo,dbNo);
        }
      }
    }
  }

  //  Check for read errors
  if ( packetFile.GetNumErrors() > 0 ) {
    ++numFailed;
     MSG("Dbm", Msg::kError) << "   input error on"
                             << fileName.c_str() << endl;
    }

  //Flush last table entry in log file.
  logFile.LogRec();


  MSG("Dbm", Msg::kInfo)
    << "Summary of processing of packets (validity rec + data):-" <<endl
    << "  Packets skipped (local seqNo): " << numLocal << endl
    << "  Packets skipped (duplicate):   " << numDuplicate << endl
    << "  Packets skipped (out of date): " << numOutOfDate << endl
    << "  Packets skipped (conflict):    " << numConflict  << endl
    << "  Packets failed :               " << numFailed    << endl
    << "  Packets tested (not imported): " << numTest      << endl
    << "  Packets imported :             " << numImported  << endl
    << "    including conflicts :        " << numImpConfl  << endl
    << "              updates :          " << numImpUpdate << endl;
  if ( numNewTables ) MSG("Dbm", Msg::kInfo) << endl
    << "  Tables Created :               " << numNewTables << endl;

  
  ofstream* log = logFile.GetStream();
  if ( log ) { 
    *log
      << "      Summary of processing of packets (validity rec + data):-" <<endl;
    if ( numLocal ) *log << "        Packets skipped (local seqNo): " 
                         << numLocal << endl;
    if ( numDuplicate ) *log << "        Packets skipped (duplicate):   " 
                 << numDuplicate << endl;
    if ( numOutOfDate ) *log << "        Packets skipped (out of date): " 
                 << numOutOfDate << endl;
    if ( numConflict ) *log << "        Packets skipped (conflict):    " 
                 << numConflict  << endl;
    if ( numFailed ) *log << "        Packets failed :               " 
                 << numFailed    << endl;
    if ( numTest ) *log << "        Packets tested (not imported): " 
                 << numTest      << endl;
    if ( numImported ) *log << "        Packets imported :             " 
                 << numImported  << endl;
    if ( numImpConfl) *log << "          including conflicts :        " 
                 << numImpConfl  << endl;
    if ( numImpUpdate ) *log << "                    updates :          " 
                   << numImpUpdate << endl;
    if ( numNewTables ) *log << endl
      << "        Tables Created :               " << numNewTables << endl;
    if ( numConflict + numFailed > 0 ) {
      *log << "      Import failed!!"  << endl;
    }
    else {
      *log << "      Import O.K."  << endl;
    }
  }

//  Close import file (even though it will as it goes out of scope).
  import.close();
    

// Exit if in Batch mode and errors have occurred.
  if (      batch
            && ( numConflict +  numFailed ) > 0  ) this->Quit();
  return packetFile.GetNumErrors() ? kFALSE : kTRUE;

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

Bool_t DbmModule::LogEntry(JobCommand *cmd) {
//
//
//  Purpose: Write log entry.
//
//  Arguments: 
//    cmd          in    Command parameters.          
//
//  Return:    kTRUE if output successful.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Write database log entry using information supplied.

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

// None.

// Process all options.

  std::list<std::string> defaultOpts;
  defaultOpts.push_back("--Table");
  defaultOpts.push_back("--Reason");
  defaultOpts.push_back("--SeqNoMin");
  defaultOpts.push_back("--Detector");
  defaultOpts.push_back("--SimFlag");

  DbmCmdOptions opts("--Table:"
                     "--Reason:"
                     "--SeqNoMin:"
                     "--Detector:"
                     "--SimFlag:"
                     "--SeqNoMax:"
                     "--NumSeqNo:"
                     "--Task:"
                     "--UpdateTime:"
                     "--DatabaseNumber",
                     cmd, &defaultOpts);
  if ( ! opts.IsValid() ) return kFALSE;


//  Deal with compulsory "options" (please excuse the contradition).

  if ( ! opts.TestOpt("--Table" ) ) {
    MSG("Dbm",Msg::kWarning) << "Table name not specified." << endl;
    return kFALSE;
  }
  string tableName = opts.GetOptString("--Table");

  if ( ! opts.TestOpt("--Reason" ) ) {
    MSG("Dbm",Msg::kWarning) << "Reason not specified." << endl;
    return kFALSE;
  }
  string reason = opts.GetOptString("--Reason");

  if ( ! opts.TestOpt("--SeqNoMin" ) ) {
    MSG("Dbm",Msg::kWarning) << "SeqNoMin not specified." << endl;
    return kFALSE;
  }
  Int_t logSeqNoMin = opts.GetOptInt("--SeqNoMin");

// Deal with optional options.

  UInt_t dbNo = 0;
  if ( opts.TestOpt("--DatabaseNumber") ) dbNo = opts.GetOptInt("--DatabaseNumber");

// Have a quick check to see if table exists.
  int dbNoFound =   DbiTableProxyRegistry::Instance().GetCascader().GetTableDbNo(tableName);
  if ( dbNoFound < 0 || dbNoFound != (int) dbNo ) {
    MSG("Dbm",Msg::kWarning) << "Sorry, cannot find table " << tableName << endl;
    return kFALSE;
  }

// If the data is already in the database, retrieve detMask, simMask and task
// and use as defaults.

  Int_t detMask  = 2;
  Int_t simMask  = 1;
  Dbi::Task task = 0;
 
  DbiResultPtr<DbiConfigSet> dataPtr(tableName,logSeqNoMin,dbNo);
  const DbiValidityRec* vrec = dataPtr.GetValidityRec();
  if ( dataPtr.GetNumRows() > 0 && vrec ) {
    MSG("Dbi",Msg::kInfo) << "Found " << tableName << " " << logSeqNoMin << endl;
    const VldRange& vrng = vrec->GetVldRange();
    detMask  = vrng.GetDetectorMask();
    simMask  = vrng.GetSimMask();
    task     = vrec->GetTask();  
  }

  if ( opts.TestOpt("--Detector") ) detMask = opts.GetOptInt("--Detector");

  if ( opts.TestOpt("--SimFlag") ) simMask = opts.GetOptInt("--SimFlag");

  Int_t logSeqNoMax = logSeqNoMin;
  if ( opts.TestOpt("--SeqNoMax") ) logSeqNoMax = opts.GetOptInt("--SeqNoMax");

  Int_t logNumSeqNo = logSeqNoMax - logSeqNoMin + 1;
  if ( opts.TestOpt("--NumSeqNo") ) logNumSeqNo = opts.GetOptInt("--NumSeqNo");

  if ( opts.TestOpt("--Task") ) task = opts.GetOptInt("--Task");

  VldTimeStamp updateTime;
  if ( opts.TestOpt("--UpdateTime") 
     ) updateTime = Dbi::MakeTimeStamp(opts.GetOptString("--UpdateTime"));

// Create a DbiLogEntry, ask for confirmation and write it out.
  DbiLogEntry logEntry(tableName,reason,detMask,simMask,task,
                       logSeqNoMin,logSeqNoMax,logNumSeqNo);
  logEntry.SetUpdateTime(updateTime);
  MSG("Dbi",Msg::kInfo) << "Created:-\n " 
                        <<  logEntry << endl;
  char ans = '?';
  while ( ans != 'y' && ans != 'n' ) {
    cout << "O.K. to output ? Ans y or n: ";
    cin  >> ans;
  }

  if ( ans == 'y' ) {
    bool ioResult = logEntry.Write(dbNo);
    if ( ioResult ) MSG("Dbi",Msg::kInfo) << "Successfully wrote entry"
                                          << endl;
    return ioResult;
  }
  MSG("Dbi",Msg::kInfo) << "Entry not written" << endl;
  return kFALSE;

}

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

void DbmModule::ListTables(const DbmNameFilter& nameFilter,
                           UInt_t dbNo,
                           std::list<string>& tableNames) const {
//
//
//  Purpose:  Assemble list of tables matching filter
//
//  Arguments: 
//    
//    nameFilter   in      Filter string
//    dbNo         in      Cascade number of Database to be used.
//    tableNames   in/out  List to hold matching names (not cleared)
//
//  Contact:   N. West
//

 //  Loop over all candidate tables (that have a corresponding
 // *Validity table).
  auto_ptr<DbiStatement> stmtDb(fCascader.CreateStatement(dbNo));
  if ( ! stmtDb.get() ) {
    MSG("Dbm", Msg::kInfo) << "Cascade entry " << dbNo
            << " empty! " << endl;
    return; 
  }
  TSQLStatement* stmt = stmtDb->ExecuteQuery("show tables");
  if ( ! stmt ) {
    MSG("Dbm", Msg::kInfo) << "Cannot find any tables! " << endl;
    return; 
  }

  while ( kTRUE ) {
    TString name;
    if ( ! stmt->NextResultRow() ) break;
    name = stmt->GetString(0);
    string tableName = name.Data();
    MSG("Dbm", Msg::kDebug )<< "examining table " << tableName << endl;

    if (    tableName.size() > 3
         && tableName.substr(tableName.size()-3) == "VLD" ) {
      tableName.erase(tableName.size()-3);
      if ( nameFilter.Test(tableName) ){
        tableNames.push_back(tableName);
        MSG("Dbm", Msg::kInfo) << "preparing to export: " << tableName << endl;
      }
    }
  }
  delete stmt;

  if(tableNames.size()==0)
        MSG("Dbm", Msg::kInfo) << "error no tables found! " << endl;
}

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

Bool_t DbmModule::Query(JobCommand *cmd) {
//
//
//  Purpose:  Simulate a validity query and display summary of results.
//
//  Arguments: 
//    cmd          in    Command parameters.          
//
//  Return: kTRUE if query satisfied.
//
//  Contact:   N. West
//

  Bool_t satisfied = kFALSE;

  // Establish defaults.
  string pattern           = "*";
  Detector::Detector_t det = Detector::kFar;
  SimFlag::SimFlag_t   sim = SimFlag::kData;
  VldTimeStamp date;
  Dbi::Task task = 0;

  // Replace as required with supplied parameters.

  DbmCmdOptions opts("--Table:--Detector:--Date:--SimFlag:--Task", cmd);
  if ( ! opts.IsValid() ) return kFALSE;
   
  if ( opts.TestOpt("--Table") ) pattern = opts.GetOptString("--Table");
  if ( opts.TestOpt("--Detector") ) {
    string detstr = opts.GetOptString("--Detector");
    if ( detstr == "Unknown"   ) det = Detector::kUnknown;
    if ( detstr == "Near"      ) det = Detector::kNear;
    if ( detstr == "Far"       ) det = Detector::kFar;
    if ( detstr == "CalDet"    ) det = Detector::kCalDet;
    if ( detstr == "Calib"     ) det = Detector::kCalib;
    if ( detstr == "TestStand" ) det = Detector::kTestStand;
    if ( detstr == "Mapper"    ) det = Detector::kMapper;
  }
  if ( opts.TestOpt("--SimFlag") ) {
    string simstr = opts.GetOptString("--SimFlag");
    if ( simstr == "Unknown"     ) sim = SimFlag::kUnknown;
    if ( simstr == "Data"        ) sim = SimFlag::kData;
    if ( simstr == "DaqFakeData" ) sim = SimFlag::kDaqFakeData;
    if ( simstr == "MC"          ) sim = SimFlag::kMC;
    if ( simstr == "Reroot"      ) sim = SimFlag::kReroot;
  }
  if ( opts.TestOpt("--Date") ) {
    string datestr = opts.GetOptString("--Date");
    date = Dbi::MakeTimeStamp(datestr);
  }

  if ( opts.TestOpt("--Task") ) {
   task = opts.GetOptInt("--Task");
  }

  //  Create context for query.
  VldContext vc(det,sim,date);
  
  //  Establish list of names.
  list<string> tableNames;
  this->ListTables(pattern,0,tableNames);

  //  Loop over all tables applying query.

  for ( list<string>::const_iterator itr = tableNames.begin();
        itr != tableNames.end();
        ++itr ) {
    DbiResultPtr<DbiConfigSet> rptr(*itr,vc,task);
    MSG("Dbm", Msg::kInfo) << "Query for table: " << *itr ;
    if ( ! rptr.GetNumRows() ) MSG("Dbm",Msg::kInfo) << " failed\n";  
    else {
      MSG("Dbm",Msg::kInfo) << " succeeded. " 
              << rptr.GetNumRows() << " rows"
              << "\n  ResultKey: " << *rptr.GetKey();
      satisfied = kTRUE;
    }  
    MSG("Dbm", Msg::kInfo) << "  ValidityRec: " << *rptr.GetValidityRec() << endl;
    
  }

  return satisfied;  // but not smug!


}

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

void DbmModule::Quit() const {
    MSG("Dbm", Msg::kInfo)  << "Exit forced due to above errors" << endl;
    exit(1);
}

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

void DbmModule::SetLogLevel(JobCommand *cmd) const {
//
//
//  Purpose:  Set Log Level
//
//  Arguments: 
//    cmd          in    Command parameters.          
//

  const char* stream = cmd->PopOpt();
  const char* level  = cmd->PopOpt();
  if ( ! stream or ! level ) {
     MSG("Dbm",Msg::kWarning) 
       << "Please supply both a stream and a level to the SetLogLevel command\n" << endl;
     return;
  }

  bool set_ok = Util::MsgLevel(stream,level);
  MSG("Dbm",Msg::kInfo) << "Setting MessageService level on stream " << stream
                  << " to level " << level
                  << ( set_ok ? " succeeded" : " failed" ) << endl;
  
}
//.....................................................................

void DbmModule::SetTimer(JobCommand *cmd) const {
//
//
//  Purpose:  Set Timer on/off
//
//  Arguments: 
//    cmd          in    Command parameters.          
//

  bool enableTimer = true;
  const char* mode = cmd->PopOpt();
  if ( mode and ! strcmp(mode,"off") ) enableTimer = false;
  DbiTimerManager::gTimerManager.Enable(enableTimer);
  MSG("Dbm",Msg::kInfo) << "Setting Timer " << (enableTimer ? "on" : "off") << endl;
  
}

/*    Template for New Member Function

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

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

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

//  None.


}

*/

---