TDbiCascader.cxx

Go to the documentation of this file.

//////////////////////////////////////////////////////////////////////////
////////////////////////////     ROOT API     ////////////////////////////
//////////////////////////////////////////////////////////////////////////


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

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

#include "TDbi.hxx"
#include "TDbiCascader.hxx"
#include "TDbiString.hxx"
#include <TSK_DBI_Log.hxx>
#include <MsgFormat.h>
using std::endl;
#include "UtilString.hxx"
#include <TSK_DBI_Log.hxx>

ClassImp(TDbiCascader)

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


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

//.....................................................................
///\verbatim
///
///  Purpose:  Default constructor
///
///  Arguments: beQuiet - If true, fail quietly. If not be more verbose.
///
///  Return:    n/a
///
///  Throws:  ENoEnvironment(); if there is no environment defined
///               throw EBadDatabase(); if any other error occured
///  Contact:   N. West
///
///  Specification:-
///  =============
///
///  o  Create Cascader.
///
///
///  Program Notes:-
///  =============
///
///  Current cascader configuration comes from 3 environmental
///  variables:-
///
///    SK_TSQL_URL             a semi-colon separated list of URLs
///    SK_TSQL_USER            user name (one or a semi-colon separated list)
///    SK_TSQL_PSWD            password (one or a semi-colon separated list)
///
///    or the _UPDATE alternatives e.g. SK_TSQL_UPDATE_USER
///
/// The _UPDATE versions take priority.
///\endverbatim
TDbiCascader::TDbiCascader(bool beQuiet):
fGlobalSeqNoDbNo(-1)
{


  SK_DBI_Trace( "Creating TDbiCascader" << "  ");

// Extract args from  SK_TSQL environmental variables
  const char*      strUser = gSystem->Getenv("SK_TSQL_UPDATE_USER");
  if ( ! strUser ) strUser = gSystem->Getenv("SK_TSQL_USER");
  const char*      strPswd = gSystem->Getenv("SK_TSQL_UPDATE_PSWD");
  if ( ! strPswd ) strPswd = gSystem->Getenv("SK_TSQL_PSWD");
  const char*       strUrl = gSystem->Getenv("SK_TSQL_UPDATE_URL");
  if ( !strUrl )    strUrl = gSystem->Getenv("SK_TSQL_URL");
  string userList     = ( strUser ) ? strUser : "";
  string pswdList     = ( strPswd ) ? strPswd : "";
  string urlList      = ( strUrl  ) ? strUrl  : "";

  
  if ( urlList == "" || userList == "" || pswdList == "" ) {
    /*
       If beQuiet is true be mute about why we are failing, leave it to the caller to deal with.
    */
    if(!beQuiet)
    {
       std::cout<<"error!"<<std::endl;
        SK_DBI_Severe("Cannnot open a Database cascade;\n"
        << "   the environmental variables SK_TSQL_*:-" << endl
        << "USER: \"" << userList << "\" PSWD:\"" << pswdList
        << "\" URL:\"" << urlList << endl
        << " are either not defined or empty.\n"
        << "   Please check your settings of SK_TSQL_USER,"
                << " SK_TSQL_PSWD and SK_TSQL_URL\n" );

    throw ENoEnvironment();
   }
  }

  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 = "";

    std::cout << "Make connection " << url << " " << user << " " << pswd << std::endl;
    TDbiConnection* con;
    // If we are testing the cascade for validity, just try connecting once, otherwise use defaults.
    if(!beQuiet)
      con  = new TDbiConnection(url,user,pswd,1);
    else
      con  = new TDbiConnection(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<TDbiStatement>  stmtDb(new TDbiStatement(*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 ) {
      if(!beQuiet)  SK_DBI_Severe("  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;
    }
  }

  SK_DBI_Info( *this);

//  Abort, if there have been any failures.
  if ( fail ) {
    throw EBadDatabase();
  }

}

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

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


  SK_DBI_Trace( "Destroying TDbiCascader" << "  ");

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

}

//.....................................................................
///\verbatim
///
///  Purpose:  Output TDbiCascader 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 TDbiCascader status to ostream.
///
///  Program Notes:-
///  =============
///
///  None.
///\endverbatim
ostream& operator<<(ostream& os, const TDbiCascader& cascader) {


  os << "TDbiCascader 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;

}
 bool TDbiCascader::canConnect()
{
  SK_DBI_Trace(" bool TDbiCascader::canConnect() started");

  try
    {
      TDbiCascader test(false);
      //   test.NOOP();
      SK_DBI_Trace(" bool TDbiCascader::canConnect() return true");
      return true;
    }
  catch(...)
    {
      SK_DBI_Trace(" bool TDbiCascader::canConnect() return false");
      return false;
    }
}
//.....................................................................
///\verbatim
///
///  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!).
///\endverbatim
Int_t TDbiCascader::AllocateSeqNo(const string& tableName,
                                 Int_t requireGlobal, /* =0 */
                                 Int_t dbNo  /* = 0 */) const {



  bool isTemporary = IsTemporaryTable(tableName,dbNo);

  //  Deal with global requests.

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

  // Deal with local requests

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

}

//.....................................................................
///\verbatim
///
///  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<TDbiStatement> stmt(cascader.CreateStatement(dbNo));
///\endverbatim
TDbiStatement* TDbiCascader::CreateStatement(UInt_t dbNo) const {


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

}
//.....................................................................
///\verbatim
///
///  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.
///
///\endverbatim
Int_t TDbiCascader::CreateTemporaryTable(const string& tableNameMc,
                                       const string& tableDescr) {


//  Check that input args look plausible.

  string tableName = UtilString::ToUpper(tableNameMc);
  if (    tableName == ""
       || tableDescr[0] != '('
       || tableDescr[tableDescr.size()-1] != ')' ) {
       SK_DBI_Severe( "Illegal input args:-" << "  "
                           << "     Table Name: " << tableName
                           << "  Table Description: " << tableDescr
                           <<"  ");
    return -1;
  }

// Find a DB that will accept the command.
  string sqlMakeTable;

  Int_t dbNoAcc       = -1;
  auto_ptr<TDbiStatement> 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;
      }
        stmtDb->PrintExceptions();
   }
  }

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

// Make connection permanent if not already.
  TDbiConnection& conDb = *fConnections[dbNoAcc];
  if ( conDb.IsTemporary() ) {
    conDb.SetPermanent();
    SK_DBI_Info( "Making connection: " << conDb.GetUrl()
                           << " permanent to preserve temporary tables." << "  ");
  }

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

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

}

//.....................................................................
///
///
///  Purpose:  Return a connection to caller (TDbiCascader retains ownership)
///
const TDbiConnection* TDbiCascader::GetConnection(UInt_t dbNo) const{


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

}
//.....................................................................
///
///
///  Purpose:  Return a connection to caller (TDbiCascader retains ownership)

TDbiConnection* TDbiCascader::GetConnection(UInt_t dbNo) {


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

}


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

  string dbName;

  if ( dbNo < this->GetNumDb() ) dbName = fConnections[dbNo]->GetDbName();
  else SK_DBI_Warn( "Database does not contain entry " << dbNo << "  ");
  return dbName;

}

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


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

  SK_DBI_Warn( "Database does not contain entry " << dbName << "  ");
  return -1;

}

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


  Int_t status = GetStatus(dbNo);

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

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


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

// If table name has any lower case letters then fail.
  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 TDbiConnection* con =  this->GetConnection(dbNoTry);
    if ( con && con->TableExists(tableName) ) return dbNoTry;
  }

  return -1;

}
//.....................................................................
///\verbatim
///
///  Purpose: Hold temporary connections open
///
///  Specification:-
///  =============
///
///  Hold all connections open by telling them that they have a
///  connected statement.
///
///  Program Notes:-
///  =============
///
///  See TDbiConnectionMaintainer for use.
///\endverbatim
void TDbiCascader::HoldConnections() {


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

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


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

}


// Private Locker object

//.....................................................................
///\verbatim
///
///  Purpose:  Ctor: Create a lock on a table accessed via connection.
///            (will be released by dtor)
///  Arguments:
///          stmtDB    in  TDbiStatement (given to Lock).
///\endverbatim
TDbiCascader::Lock::Lock(TDbiStatement* stmtDB, const string& seqnoTable, const string& dataTable) :
fStmt(stmtDB),
fSeqnoTableName(seqnoTable),
fDataTableName(dataTable),
fLocked(kFALSE)
{


  if ( ! fStmt ) {
       SK_DBI_Severe( "Cannot obtain statment to set lock" << "  ");
    return;
  }

  this->SetLock(kTRUE);

}
//.....................................................................
///  Purpose:  Dtor: Clear lock
TDbiCascader::Lock::~Lock() {
//
//


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

}

//.....................................................................
///\verbatim
///
///  Purpose:  Set or clear lock.
///
///  Arguments:
///    setting      in    Required setting of lock. [default kTRUE]
///
///  Program Notes:-
///  =============
///
///  No-op if locked otherwise use the LOCK TABLES command.
///\endverbatim

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


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

  string sql;

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

}
//.....................................................................
///\verbatim
///
///  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 TDbiConnectionMaintainer for use.
///\endverbatim
void TDbiCascader::ReleaseConnections() {


  for (UInt_t dbNo = 0; dbNo < fConnections.size(); ++dbNo )
    fConnections[dbNo]->DisConnectStatement();
}
//.....................................................................
///\verbatim
///
///  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.
///\endverbatim
Int_t TDbiCascader::ReserveNextSeqNo(const string& tableName,
                                    Bool_t isGlobal,
                                    UInt_t dbNo) const {


  TDbiString sql;

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

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

  // Check that required SEQNO table exists.

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

      SK_DBI_Log( "Database: " << dbNo
                             << " create local SEQNO table query: " << sql.c_str() << "  ");
      stmtDb->ExecuteUpdate(sql.c_str());
      if ( stmtDb->PrintExceptions() ) return 0;
      sql.Clear();
      sql << "INSERT INTO " <<  seqnoTableName << " VALUES ('*',0)";
      SK_DBI_Log( "Database: " << dbNo
                             << " prime local SEQNO table query: " << sql.c_str() << "  ");
      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() ) {
       SK_DBI_Severe( "Unable to lock " << seqnoTableName << "  ");
    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";
  SK_DBI_Log( " query: " << sql.c_str() << "  ");
  TSQLStatement* stmt = stmtDb->ExecuteQuery(sql.c_str());

  Int_t seqNoDefault = 0;
  if ( stmt && stmt->NextResultRow() ) {
    seqNoDefault = stmt->GetInt(1);
  }
  else {
       SK_DBI_Severe( "Unable to find default SeqNo"
                           << " due to above error" << "  ");
    delete stmt;
    stmt = 0;
    return 0;
  }
  Int_t seqNoTable = seqNoDefault;
  if ( stmt->NextResultRow() ) {
    seqNoTable = stmt->GetInt(1);
  }
  delete stmt;
  stmt = 0;
  SK_DBI_Log( "  query returned last used seqno: " << seqNoTable << "  ");

//  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 + TDbi::kMAXLOCALSEQNO;
    sql.Clear();
    sql << "select max(SEQNO) from " << tableName << "VLD"
        << " where SEQNO between " << seqNoMin << " and " << seqNoMax;
    SK_DBI_Log( "Database: " << dbNo
                         << " max  SEQNO query: " << sql.c_str() << "  ");
    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 ) {
         SK_DBI_Severe(  "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 << "  ");
        seqNoTable = minValue;
    }
  }


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

  seqNoTable++;

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

  return seqNoTable;

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


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

}