DbiCascader.cxx

Go to the documentation of this file.

#include <cstdlib>
#include <memory>
using std::auto_ptr;
#include <sstream>

#include "TList.h"
#include "TROOT.h"
#include "TSQLStatement.h"
#include "TSystem.h"

#include "DatabaseInterface/Dbi.h"
#include "DatabaseInterface/DbiCascader.h"
#include "DatabaseInterface/DbiString.h"
#include "LeakChecker/Lea.h"
#include "MessageService/MsgService.h"
#include "Util/UtilString.h"

ClassImp(DbiCascader)

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

CVSID("$Id: DbiCascader.cxx,v 1.73 2009/04/18 08:12:37 west Exp $");

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

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

DbiCascader::DbiCascader():
fGlobalSeqNoDbNo(-1)
{
//
//
//  Purpose:  Default constructor
//
//  Arguments:
//
//  Return:    n/a
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o  Create Cascader.


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

//  Current cascader configuration comes from 3 environmental
//  variables:-
//
//    ENV_TSQL_URL             a semi-colon separated list of URLs
//    ENV_TSQL_USER            user name (one or a semi-colon separated list)
//    ENV_TSQL_PSWD            password (one or a semi-colon separated list)
//   
//    or the _UPDATE alternatives e.g. ENV_TSQL_UPDATE_USER
//   
// The _UPDATE versions take priority.

  LEA_CTOR    //Leak Checker

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

// Extract args from  ENV_TSQL environmental variables
  const char*      strUser = gSystem->Getenv("ENV_TSQL_UPDATE_USER");
  if ( ! strUser ) strUser = gSystem->Getenv("ENV_TSQL_USER");
  const char*      strPswd = gSystem->Getenv("ENV_TSQL_UPDATE_PSWD");
  if ( ! strPswd ) strPswd = gSystem->Getenv("ENV_TSQL_PSWD");
  const char*       strUrl = gSystem->Getenv("ENV_TSQL_UPDATE_URL");
  if ( !strUrl )    strUrl = gSystem->Getenv("ENV_TSQL_URL");
  string userList     = ( strUser ) ? strUser : "";
  string pswdList     = ( strPswd ) ? strPswd : "";
  string urlList      = ( strUrl  ) ? strUrl  : "";

  if ( urlList == "" || userList == "" || pswdList == "" ) {
     MSG("Dbi",Msg::kFatal) 
        << "Cannnot open a Database cascade;\n"
        << "   the environmental variables ENV_TSQL_*:-" << endl
        << "USER: \"" << userList << "\" PSWD:\"" << pswdList 
        << "\" URL:\"" << urlList << endl
        << " are either not defined or empty.\n"
        << "   Please check your settings of ENV_TSQL_USER,"
        << " ENV_TSQL_PSWD and ENV_TSQL_URL\n" 
        << "Aborting due to above errors" << endl;
      exit(1);
  }

  std::vector<std::string> users, pswds, urls;
  UtilString::StringTok(users, userList, ";");
  UtilString::StringTok(pswds, pswdList, ";");
  UtilString::StringTok(urls,  urlList,  ";");

  bool fail = false;

  for (unsigned entry = 0; entry < urls.size(); ++entry ) { 
    string url  = urls[entry];
    string user = ( entry >= users.size() ) ? users[0] : users[entry];
    string pswd = ( entry >= pswds.size() ) ? pswds[0] : pswds[entry];

    // Handle empty password designated as '\0' (an empty null terminated character string)
    if ( pswd == "\\0" ) pswd = "";

    DbiConnection* con = new DbiConnection(url,user,pswd);
    fConnections.push_back(con);
    if ( ! con->Open() ) {
      fail = true;
      continue;
    }

//  Attempt to locate first GlobalSeqNo/GLOBALSEQNO table.
    if ( fGlobalSeqNoDbNo != -1 ) continue;
    auto_ptr<DbiStatement>  stmtDb(new DbiStatement(*con));
    if ( ! stmtDb.get() ) continue;
    TSQLStatement* stmt = stmtDb->ExecuteQuery("Select * from GLOBALSEQNO where 1=0");
    if ( stmt ) {
      fGlobalSeqNoDbNo = fConnections.size()-1;
      delete stmt;
      stmt = 0;
    }
   
//  Check for presence of a DBI_STATE_FLAG table

    if ( this->GetTableDbNo("DBI_STATE_FLAGS",entry) != -1 ) {
      MSG("Dbi",Msg::kFatal) << "  POSSIBLE VERSION SHEAR DETECTED !!!\n"
        << "    The DBI_STATE_FLAGS table is present on cascade entry " << entry << ".  This table will\n"
        << "    only be introduced to manage backward incompatible changes that could lead\n"
        << "    to version shear between the code and the database.  This version of the\n"
        << "    code does not support the change the presence of that table indicates\n"
        << "    so has to shut down. \n";
        fail = true;
    }
  }

  MSG("Dbi",Msg::kInfo) << *this;
 
//  Abort, if there have been any failures.
  if ( fail ) {
    MSG("Dbi",Msg::kFatal)
        << "Aborting due to above errors" << endl;
    exit(1);
  }

}

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

DbiCascader::~DbiCascader() {
//
//
//  Purpose: Destructor
//

  LEA_DTOR    //Leak Checker

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

  for (Int_t dbNo = this->GetNumDb()-1; dbNo >= 0; --dbNo) delete fConnections[dbNo];

}

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

ostream& operator<<(ostream& os, const DbiCascader& cascader) {
//
//
//  Purpose:  Output DbiCascader status to message stream.
//
//  Arguments: 
//    os           in    ostream to output on
//    cascader     in    Cascader to be output
//
//  Return:        ostream   
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Output DbiCascader status to ostream.

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

//  None.

  os << "DbiCascader Status:- " << endl
     << "Status   URL" << endl << endl;

  int maxDb = cascader.GetNumDb();
  for (Int_t dbNo = 0; dbNo < maxDb; ++dbNo)
      os << cascader.GetStatusAsString(dbNo) << " "
         << ( ( dbNo == cascader.fGlobalSeqNoDbNo ) ? "(auth)  " : "        ")
         << cascader.GetURL(dbNo) << endl;
  os << endl;
  return os;

}

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

Int_t DbiCascader::AllocateSeqNo(const string& tableName,
                                 Int_t requireGlobal, /* =0 */
                                 Int_t dbNo  /* = 0 */) const {
//
//
//  Purpose:  Allocate a unique (either locally or globally) SEQNO.
//
//  Arguments: 
//   tableName       in    The table for which the SEQNO is required.
//   requireGlobal   in    The type of SEQNO required:-
//                           > 0  Must be global
//                           = 0  Must be global if supplied dbNo is authorising
//                                and table isn't temporary otherwise local
//                           < 0  Must be local
//   dbNo            in     The entry in the cascade for which the SEQNO is required
// 
//  Return:    The allocated SEQNO or 0 if failure.
//
//  Contact:   N. West

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

//  Requests for SEQNOs take account of any pre-existing entries; local entries
//  should only be used for development and this allows for the LOCALSEQNO table 
//  and the local data to be wiped at different times without causing conflicts.
//  Global entries should not be a problem as the GLOBALSEQNO table isn't wiped
//  but it provides protection in case the table is damaged (which has happened!).

  bool isTemporary = IsTemporaryTable(tableName,dbNo);

  //  Deal with global requests.

  if (     requireGlobal > 0
      || ( requireGlobal == 0 && dbNo == fGlobalSeqNoDbNo && ! isTemporary ) ) {
    if ( fGlobalSeqNoDbNo < 0 ) {
      MSG("Dbi",Msg::kWarning) << "Unable to issue global SEQNO - no authorising DB in cascade\n"
                               << "  will issue local one instead" << endl;
    }
    else if ( isTemporary ) {
      MSG("Dbi",Msg::kWarning) << "Unable to issue global SEQNO - " << tableName << " is temporary\n"
                               << "  will issue local one instead" << endl;
    }
    else return this->ReserveNextSeqNo(tableName,true,fGlobalSeqNoDbNo);
  }

  // Deal with local requests

  return this->ReserveNextSeqNo(tableName,false,dbNo);

}

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

DbiStatement* DbiCascader::CreateStatement(UInt_t dbNo) const {
//
//
//  Purpose:  Return a Statement to caller.
//
//  Arguments: 
//    dbNo     in    Database no. in cascade for which statement
//                   is required.
//
//  Return:    Created statement (or 0 if creation failed). 
//             User must delete.
//  Program Notes:-
//  =============

//  As the caller is responsible for destroying the statement after use
//  consider:-
//  
//  #include <memory>
//  using std::auto_ptr;
//  
//  ...
//  
//  auto_ptr<DbiStatement> stmt(cascader.CreateStatement(dbNo));


  if ( this->GetStatus(dbNo) == kFailed ) return 0;
  DbiConnection& conDb = *fConnections[dbNo];
  DbiStatement* stmtDb = new DbiStatement(conDb);
  stmtDb->PrintExceptions();
  return stmtDb;

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

Int_t DbiCascader::CreateTemporaryTable(const string& tableNameMc,
                                       const string& tableDescr) {
//
//
//  Purpose:  Creat temporary table with associated validity table
//
//  Arguments: 
//    tableNameMc  in    Table name
//    tableDescr   in    Table description as parenthesised comma 
//                       separated list e.g.:-
//                       "(MyInt int, MyFloat float, MyString text)"
//
//  Return:    The database cascade number on which the table was
//             created or = -1 if unable to create.
//
//  Contact:   N. West
//
//  Specification:-
//  =============
//
//  o Loop over all databases in cascade, starting at the highest
//    priority i.e. entry 0 and find the first that will accept create 
//    temporary table requests.  Return -1 if none will.
//
//  o Make connection permanent so that temporary data won't be lost.
//
//  o Generate and submit requests to create temporary table with
//    associated validity table and return the cascade number of
//    the database that has accepted the tables, recording table
//    name in fTemporaryTables.
//

  
//  Check that input args look plausible.

  string tableName = UtilString::ToUpper(tableNameMc);
  if (    tableName == "" 
       || tableDescr[0] != '(' 
       || tableDescr[tableDescr.size()-1] != ')' ) {
    MAXMSG("Dbi",Msg::kError,20) << "Illegal input args:-" << endl
                           << "     Table Name: " << tableName
                           << "  Table Description: " << tableDescr
                           <<endl;
    return -1;
  }
  
// Find a DB that will accept the command.
  string sqlMakeTable;

  Int_t dbNoAcc       = -1;
  auto_ptr<DbiStatement> stmtDb; 
  for (UInt_t dbNoTry = 0; dbNoTry < fConnections.size(); ++dbNoTry ) {
    stmtDb.reset(this->CreateStatement(dbNoTry));    
    if ( stmtDb.get() ) {
      sqlMakeTable = " create temporary table ";
      sqlMakeTable += tableName;
      sqlMakeTable += " ";
      sqlMakeTable += tableDescr;
      sqlMakeTable += ";";
      stmtDb->ExecuteUpdate(sqlMakeTable.c_str());
      if ( stmtDb->GetExceptionLog().IsEmpty() ) {
        dbNoAcc = dbNoTry;
        this->GetConnection(dbNoAcc)->SetTableExists(tableName);
        break;
      }
      if ( MsgService::Instance()->IsActive("Dbi",Msg::kSynopsis) )stmtDb->PrintExceptions();
   }
  }

  if ( dbNoAcc < 0 ) {
    if ( stmtDb.get()) stmtDb->PrintExceptions();
    return -1;
  }

// Make connection permanent if not already.
  DbiConnection& conDb = *fConnections[dbNoAcc];
  if ( conDb.IsTemporary() ) {
    conDb.SetPermanent();
    MSG("Dbi",Msg::kInfo)  << "Making connection: " << conDb.GetUrl() 
                           << " permanent to preserve temporary tables." << endl;
  }

// Create SQL to create auxillary validity table and write to same Db.
  sqlMakeTable = Dbi::GetVldDescr(tableName.c_str(),true);

  MSG("Dbi",Msg::kSynopsis) << "Validity Table creation: "
                         << " Database: " << dbNoAcc << " "
                         << sqlMakeTable << endl;
  stmtDb->ExecuteUpdate(sqlMakeTable.c_str());
  if ( stmtDb->PrintExceptions() ) return -1;
  this->GetConnection(dbNoAcc)->SetTableExists(tableName+"VLD");
  fTemporaryTables[tableName] = dbNoAcc;
  return dbNoAcc;

}

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

const DbiConnection* DbiCascader::GetConnection(UInt_t dbNo) const{
//
//
//  Purpose:  Return a connection to caller (DbiCascader retains ownership)


  if ( this->GetStatus(dbNo) == kFailed ) return 0;
  return fConnections[dbNo];

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

DbiConnection* DbiCascader::GetConnection(UInt_t dbNo) {
//
//
//  Purpose:  Return a connection to caller (DbiCascader retains ownership)


  if ( this->GetStatus(dbNo) == kFailed ) return 0;
  return fConnections[dbNo];

}


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

string DbiCascader::GetDbName(UInt_t dbNo) const {
//
//
//  Purpose:  Return Database Name for cascade entry number.

  string dbName;

  if ( dbNo < this->GetNumDb() ) dbName = fConnections[dbNo]->GetDbName();
  else MSG("Dbi",Msg::kWarning) << "Database does not contain entry " << dbNo << endl;
  return dbName;

}

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

Int_t DbiCascader::GetDbNo(const string& dbName) const {
//
//
//  Purpose:  Return number of first DB in cascade with name dbName.
//
//  Return:   Database number corresponding to dbName or -1 if none.   

  for ( unsigned dbNo = 0; dbNo < this->GetNumDb(); ++dbNo) {
    if ( dbName == fConnections[dbNo]->GetDbName() ) return dbNo;
  }

  MSG("Dbi",Msg::kWarning) << "Database does not contain entry " << dbName << endl;
  return -1;

}

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

string DbiCascader::GetStatusAsString(UInt_t dbNo) const {
//
//
//  Purpose:  Return DB connection status as a string.
//
//  Arguments: 
//    dbNo         in    Database number (0..GetNumDb()-1)

  Int_t status = GetStatus(dbNo);

  switch ( status ) {
  case kClosed:  return "Closed";
  case kOpen:    return "Open  ";
  default:       return "Failed";
  }

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

Int_t DbiCascader::GetTableDbNo(const string& tableName, 
                                Int_t selectDbNo /* -1 */) const {
//
//
//  Purpose:  Return cascade number of first database that holds table 
//            or -1 if none.

// If selectDbNo >= 0 only look in this entry in the cascade.

// If table name has any lower case letters then fail.  
// Lower case are not allowed by ORACLE which kindly 
// converts to upper case so SQL would pass if upper case version
// exists.
  string::const_iterator itr    = tableName.begin();
  string::const_iterator itrEnd = tableName.end();
  while ( itr != itrEnd ) if ( islower(*itr++) ) return -1;

// Loop over cascade looking for table.

  for (UInt_t dbNoTry = 0; dbNoTry < fConnections.size(); ++dbNoTry ) {
    if ( selectDbNo >= 0 && (UInt_t) selectDbNo != dbNoTry ) continue;
    const DbiConnection* con =  this->GetConnection(dbNoTry);
    if ( con && con->TableExists(tableName) ) return dbNoTry;
  }

  return -1;

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

void DbiCascader::HoldConnections() {
//
//
//  Purpose: Hold temporary connections open 

//  Specification:-
//  =============

//  Hold all connections open by telling them that they have a 
//  connected statement. 

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

//  See DbiConnectionMaintainer for use.

  for (UInt_t dbNo = 0; dbNo < fConnections.size(); ++dbNo ) 
    fConnections[dbNo]->ConnectStatement();
}

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

Bool_t DbiCascader::IsTemporaryTable(const string& tableName,
                                     Int_t dbNo) const {
//
//
//  Purpose:  Return kTRUE if tableName is temporary in cascade member dbNo

  map<string,Int_t>::const_iterator itr
                                     = fTemporaryTables.find(tableName);
  return (     itr != fTemporaryTables.end() 
           &&  (*itr).second == dbNo );

}


// Private Locker object

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

DbiCascader::Lock::Lock(DbiStatement* stmtDB, const string& seqnoTable, const string& dataTable) :
fStmt(stmtDB),
fSeqnoTableName(seqnoTable),
fDataTableName(dataTable),
fLocked(kFALSE)
{
//
//
//  Purpose:  Ctor: Create a lock on a table accessed via connection.
//            (will be released by dtor)
//  Arguments: 
//          stmtDB    in  DbiStatement (given to Lock).
  
  LEA_CTOR_NM("DbiCascader::Lock",this);

  if ( ! fStmt ) {
    MAXMSG("Dbi",Msg::kError,20) << "Cannot obtain statment to set lock" << endl;
    return;
  }

  this->SetLock(kTRUE);

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

DbiCascader::Lock::~Lock() {
//
//
//  Purpose:  Dtor: Clear lock

  LEA_DTOR_NM("DbiCascader::Lock",this);

  this->SetLock(kFALSE);
  delete fStmt;
  fStmt = 0; 

}

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

 void DbiCascader::Lock::SetLock(Bool_t setting) {

//
//
//  Purpose:  Set or clear lock.
//
//  Arguments: 
//    setting      in    Required setting of lock. [default kTRUE]

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

//  No-op unless connect to MySQL DB, in which case it uses the LOCK TABLES command.

  if ( setting == fLocked || ! fStmt ) return;

  if ( fStmt->GetDBType() != Dbi::kMySQL ) { 
    fLocked = setting;
    return;
  }

  string sql;

  if ( setting ) {
    sql = "LOCK TABLES ";
    sql += fSeqnoTableName + " WRITE";
    if ( fDataTableName != "" ) sql += ", " + fDataTableName  + "VLD WRITE";
  }
  else {
    sql = "UNLOCK TABLES;";
  }
  MSG("Dbi",Msg::kSynopsis) << "Lock requested: " << setting 
                         << " issuing lock command: " << sql << endl;
  fStmt->ExecuteUpdate(sql.c_str());
  if ( fStmt->GetExceptionLog().IsEmpty() ) fLocked = setting;
  fStmt->PrintExceptions();

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

void DbiCascader::ReleaseConnections() {
//
//
//  Purpose: Release temporary connections held open by HoldConnections. 


//  Specification:-
//  =============

//  Undo HoldConnections() by telling all connections that they no longer
// have a connected statement.

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

//  See DbiConnectionMaintainer for use.

  for (UInt_t dbNo = 0; dbNo < fConnections.size(); ++dbNo ) 
    fConnections[dbNo]->DisConnectStatement();
}
//.....................................................................

Int_t DbiCascader::ReserveNextSeqNo(const string& tableName,
                                    Bool_t isGlobal,
                                    UInt_t dbNo) const {
//
//
//  Purpose:  Reserve the next higher available unique (either locally or globally) SEQNO.
//            in the appropriate SEQNO table.
//
//  Arguments: 
//   tableName       in    The table for which the SEQNO is required.
//   isGlobal        in    = true - reserve in GLOBALSEQNO table(dbNo must be authorizing)
//                         = false - reserve in LOCALSEQNO table (creating if required)
//   dbNo            in    The entry in the cascade holding the SEQNO table.
//
//  Return:    The allocated SEQNO or 0 if failure.
//
//  Contact:   N. West

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

//  Requests for local SEQNOs may result in the creation of a LOCALSEQNO table.

  DbiString sql;

  string seqnoTableName = isGlobal ? "GLOBALSEQNO" : "LOCALSEQNO";
  bool seqnoTableNameExists = this->TableExists(seqnoTableName,dbNo);
  bool tableNameExists      = this->TableExists(tableName,dbNo);

  auto_ptr<DbiStatement> stmtDb(this->CreateStatement(dbNo) );
  if ( ! stmtDb.get() ) return 0;

  // Check that required SEQNO table exists.

  if ( isGlobal ) {
    if ( ! seqnoTableNameExists ) {
      MAXMSG("Dbi",Msg::kError,20) << "Unable to issue global SEQNO - " << dbNo
                               << " is not an authorising DB" << endl;
      return 0;
    }
  }
  else {
    if ( ! seqnoTableNameExists ) {
      sql.Clear();
      sql << "CREATE TABLE " << seqnoTableName 
          << "(TABLENAME      CHAR(64) NOT NULL PRIMARY KEY,\n"
          << " LASTUSEDSEQNO  INT )";

      MSG("Dbi",Msg::kSynopsis) << "Database: " << dbNo
                             << " create local SEQNO table query: " << sql.c_str() << endl;
      stmtDb->ExecuteUpdate(sql.c_str());
      if ( stmtDb->PrintExceptions() ) return 0;
      sql.Clear();
      sql << "INSERT INTO " <<  seqnoTableName << " VALUES ('*',0)";
      MSG("Dbi",Msg::kSynopsis) << "Database: " << dbNo
                             << " prime local SEQNO table query: " << sql.c_str() << endl;
      stmtDb->ExecuteUpdate(sql.c_str());
      if ( stmtDb->PrintExceptions() ) return 0;
    }
  }
 
// Lock seqno table by creating a lock object on the stack.
// Table will be unlocked when lock object destroyed.

  string dataTable;
// Only pass in table name if it's not temporary and exists in
// the selected DB otherwise Lock will try to lock a non-existent table.
  if (  ! this->IsTemporaryTable(tableName,dbNo)
       && tableNameExists ) dataTable = tableName;
  Lock lock(this->CreateStatement(dbNo),seqnoTableName,dataTable);
  if ( ! lock.IsLocked() ) {
    MAXMSG("Dbi",Msg::kError,20) << "Unable to lock " << seqnoTableName << endl;
    return 0;
  }

// Find row containing last used SeqNo for this table.
// Not that comparison is case insensitive.
  sql.Clear();
  sql << "select * from " << seqnoTableName << " where TABLENAME = '*' or TABLENAME = '";
  sql << tableName + "' order by TABLENAME";
  MSG("Dbi",Msg::kSynopsis) << seqnoTableName << " query: " << sql.c_str() << endl;
  TSQLStatement* stmt = stmtDb->ExecuteQuery(sql.c_str());
  stmtDb->PrintExceptions(Msg::kDebug);
  Int_t seqNoDefault = 0;
  if ( stmt && stmt->NextResultRow() ) {
    seqNoDefault = stmt->GetInt(1);
  }
  else {
    MAXMSG("Dbi",Msg::kError,20) << "Unable to find default SeqNo"
                           << " due to above error" << endl;
    delete stmt;
    stmt = 0;
    return 0;
  }
  Int_t seqNoTable = seqNoDefault;
  if ( stmt->NextResultRow() ) {
    seqNoTable = stmt->GetInt(1);
  }
  delete stmt;
  stmt = 0;
  MSG("Dbi",Msg::kSynopsis) << "  query returned last used seqno: " << seqNoTable << endl;

//  If the table exists, make sure that the seqNo hasn't already been used.
//  This is paranoia code and expensive, so only do the check once for
//  each tableName/isGlobal/dbNo combination.

  static std::string checkedCombinations;
  ostringstream combination;
  combination << ":" << tableName << isGlobal << dbNo << ":";
  bool notChecked = checkedCombinations.find(combination.str()) == std::string::npos;
  if ( notChecked ) checkedCombinations += combination.str();
  if ( tableNameExists && notChecked ) {
    Int_t seqNoMin = seqNoDefault;
    Int_t seqNoMax = seqNoDefault + Dbi::kMAXLOCALSEQNO;
    sql.Clear();
    sql << "select max(SEQNO) from " << tableName << "VLD"
        << " where SEQNO between " << seqNoMin << " and " << seqNoMax;
    MSG("Dbi",Msg::kSynopsis) << "Database: " << dbNo
                         << " max  SEQNO query: " << sql.c_str() << endl;
    stmt  =  stmtDb->ExecuteQuery(sql.c_str());
    if ( stmtDb->PrintExceptions() ) return 0;
    Int_t  minValue = 0;
    // Queries returning group function results can be null.
    if (  stmt && stmt->NextResultRow() && ! stmt->IsNull(0) ) {
      minValue = stmt->GetInt(0);
     if ( minValue <= 0 ) minValue = 0;  // Should never happen.
    }
    delete stmt;
    stmt = 0;

    if ( minValue > seqNoTable ) {
      MAXMSG("Dbi",Msg::kError,20) 
          << "Database: " << dbNo << " "
          << seqnoTableName << " has last used SEQNO of "
          << seqNoTable << " for table " << tableName
          << ",\n    but the highest SEQNO in the band " << seqNoMin << " to " << seqNoMax
          << " is " <<  minValue << " for that table\n    "
          << seqnoTableName << " is out of date! It will be updated for " << tableName << endl;
        seqNoTable = minValue;
    }
  }

  
//  Update last used SeqNo and record in table.
  sql.Clear();
  sql << "delete from " << seqnoTableName << " where TABLENAME='";
  sql << tableName + "'";
  MSG("Dbi",Msg::kSynopsis) << "SEQNO entry removal: " << sql.c_str() << endl;
  stmtDb->ExecuteUpdate(sql.c_str());
  if ( stmtDb->PrintExceptions() ) return 0;

  seqNoTable++;

  sql.Clear();
  sql << "insert into  " << seqnoTableName << " values('";
  sql << tableName + "'," << seqNoTable << ")";
  MSG("Dbi",Msg::kSynopsis) << "SEQNO entry add: " << sql.c_str() << endl;
  stmtDb->ExecuteUpdate(sql.c_str());
  if ( stmtDb->PrintExceptions() ) return 0;

  return seqNoTable;

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

void DbiCascader::SetPermanent(UInt_t dbNo,
                               Bool_t permanent /* = true */ ) {
//
//
//  Purpose: Set connection permanent. 

  if ( dbNo < fConnections.size() ) fConnections[dbNo]->SetPermanent(permanent);

}

---