//////////////////////////////////////////////////////////////////////////
////////////////////////////     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 "IDbi.hxx"
#include "IDbiCascader.hxx"
#include "IDbiString.hxx"
#include <ICOMETLog.hxx>
#include <IMsgFormat.h>
using std::endl;
#include "UtilString.hxx"
#include <ICOMETLog.hxx>

ClassImp(COMET::IDbiCascader)

//   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:  COMET::ENoEnvironment(); if there is no environment defined
///               throw COMET::EBadDatabase(); if any other error occured
///  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.
///\endverbatim
COMET::IDbiCascader::IDbiCascader(bool beQuiet):
fGlobalSeqNoDbNo(-1)
{


  COMETTrace( "Creating COMET::IDbiCascader" << "  ");

// 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 == "" ) {
    /*
       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;
        COMETSevere("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" );

    throw COMET::ENoEnvironment();
   }
  }

  std::vector<std::string> users, pswds, urls;
  COMET::UtilString::StringTok(users, userList, ";");
  COMET::UtilString::StringTok(pswds, pswdList, ";");
  COMET::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 = "";

    COMET::IDbiConnection* con;
    // If we are testing the cascade for validity, just try connecting once, otherwise use defaults.
    if(!beQuiet)
      con  = new COMET::IDbiConnection(url,user,pswd,1);
    else
      con  = new COMET::IDbiConnection(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<COMET::IDbiStatement>  stmtDb(new COMET::IDbiStatement(*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)  COMETSevere("  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;
    }
  }

  COMETInfo( *this);

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

}

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

COMET::IDbiCascader::~IDbiCascader() {
//
//
//  Purpose: Destructor
//


  COMETTrace( "Destroying COMET::IDbiCascader" << "  ");

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

}

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


  os << "COMET::IDbiCascader 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 COMET::IDbiCascader::canConnect()
{
  COMETTrace(" bool COMET::IDbiCascader::canConnect() started");

  try
    {
      COMET::IDbiCascader test(false);
      //   test.NOOP();
      COMETTrace(" bool COMET::IDbiCascader::canConnect() return true");
      return true;
    }
  catch(...)
    {
      COMETTrace(" bool COMET::IDbiCascader::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 COMET::IDbiCascader::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 ) {
      COMETWarn( "Unable to issue global SEQNO - no authorising DB in cascade\n"
             << "  will issue local one instead" << "  ");
    }
    else if ( isTemporary ) {
      COMETWarn( "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<COMET::IDbiStatement> stmt(cascader.CreateStatement(dbNo));
///\endverbatim
COMET::IDbiStatement* COMET::IDbiCascader::CreateStatement(UInt_t dbNo) const {


  if ( this->GetStatus(dbNo) == kFailed ) return 0;
  COMET::IDbiConnection& conDb = *fConnections[dbNo];
  COMET::IDbiStatement* stmtDb = new COMET::IDbiStatement(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 COMET::IDbiCascader::CreateTemporaryTable(const string& tableNameMc,
                                       const string& tableDescr) {


//  Check that input args look plausible.

  string tableName = COMET::UtilString::ToUpper(tableNameMc);
  if (    tableName == ""
       || tableDescr[0] != '('
       || tableDescr[tableDescr.size()-1] != ')' ) {
       COMETSevere( "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<COMET::IDbiStatement> 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.
  COMET::IDbiConnection& conDb = *fConnections[dbNoAcc];
  if ( conDb.IsTemporary() ) {
    conDb.SetPermanent();
    COMETInfo( "Making connection: " << conDb.GetUrl()
         << " permanent to preserve temporary tables." << "  ");
  }

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

  COMETLog( "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 (COMET::IDbiCascader retains ownership)
///
const COMET::IDbiConnection* COMET::IDbiCascader::GetConnection(UInt_t dbNo) const{


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

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

COMET::IDbiConnection* COMET::IDbiCascader::GetConnection(UInt_t dbNo) {


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

}


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

  string dbName;

  if ( dbNo < this->GetNumDb() ) dbName = fConnections[dbNo]->GetDbName();
  else COMETWarn( "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 COMET::IDbiCascader::GetDbNo(const string& dbName) const {


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

  COMETWarn( "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 COMET::IDbiCascader::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 COMET::IDbiCascader::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 COMET::IDbiConnection* 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 COMET::IDbiConnectionMaintainer for use.
///\endverbatim
void COMET::IDbiCascader::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 COMET::IDbiCascader::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  COMET::IDbiStatement (given to Lock).
///\endverbatim
COMET::IDbiCascader::Lock::Lock(COMET::IDbiStatement* stmtDB, const string& seqnoTable, const string& dataTable) :
fStmt(stmtDB),
fSeqnoTableName(seqnoTable),
fDataTableName(dataTable),
fLocked(kFALSE)
{


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

  this->SetLock(kTRUE);

}
//.....................................................................
///  Purpose:  Dtor: Clear lock
COMET::IDbiCascader::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 COMET::IDbiCascader::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;";
  }
  COMETLog( "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 COMET::IDbiConnectionMaintainer for use.
///\endverbatim
void COMET::IDbiCascader::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 COMET::IDbiCascader::ReserveNextSeqNo(const string& tableName,
                                    Bool_t isGlobal,
                                    UInt_t dbNo) const {

  COMET::IDbiString sql;

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

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

  // Check that required SEQNO table exists.

  if ( isGlobal ) {
    if ( ! seqnoTableNameExists ) {
         COMETSevere( "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 )";

      COMETLog( "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)";
      COMETLog( "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() ) {
       COMETSevere( "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";
  COMETLog( " query: " << sql.c_str() << "  ");
  TSQLStatement* stmt = stmtDb->ExecuteQuery(sql.c_str());
  stmtDb->PrintExceptions(COMET::ICOMETLog::DebugLevel );
  Int_t seqNoDefault = 0;
  if ( stmt && stmt->NextResultRow() ) {
    seqNoDefault = stmt->GetInt(1);
  }
  else {
       COMETSevere( "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;
  COMETLog( "  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 + IDbi::kMAXLOCALSEQNO;
    sql.Clear();
    sql << "select max(SEQNO) from " << tableName << "VLD"
  << " where SEQNO between " << seqNoMin << " and " << seqNoMax;
    COMETLog( "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 ) {
         COMETSevere(  "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 + "'";
  COMETLog( "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 << ")";
  COMETLog( "SEQNO entry add: " << sql.c_str() << "  ");
  stmtDb->ExecuteUpdate(sql.c_str());
  if ( stmtDb->PrintExceptions() ) return 0;

  return seqNoTable;

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


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

}