history_schema.cxx

Go to the documentation of this file.

/********************************************************************\
 
  Name:         history_schema.cxx
  Created by:   Konstantin Olchanski
 
  Contents:     Schema based MIDAS history. Available drivers:
                FileHistory: storage of data in binary files (replacement for the traditional MIDAS history)
                MysqlHistory: storage of data in MySQL database (replacement for the ODBC based SQL history)
                PgsqlHistory: storage of data in PostgreSQL database
                SqliteHistory: storage of data in SQLITE3 database (not suitable for production use)
 
\********************************************************************/
 
#undef NDEBUG // midas required assert() to be always enabled
 
#include "midas.h"
#include "msystem.h"
#include "mstrlcpy.h"
 
#include <math.h>
 
#include <vector>
#include <list>
#include <string>
#include <map>
#include <unordered_map>
#include <algorithm>
 
// make mysql/my_global.h happy - it redefines closesocket()
#undef closesocket
 
//
// benchmarks
//
// /usr/bin/time ./linux/bin/mh2sql . /ladd/iris_data2/alpha/alphacpc09-elog-history/history/121019.hst
// -rw-r--r-- 1 alpha users 161028048 Oct 19  2012 /ladd/iris_data2/alpha/alphacpc09-elog-history/history/121019.hst
// flush   10000, sync=OFF    -> 51.51user 1.51system 0:53.76elapsed 98%CPU
// flush 1000000, sync=NORMAL -> 51.83user 2.09system 1:08.37elapsed 78%CPU (flush never activated)
// flush  100000, sync=NORMAL -> 51.38user 1.94system 1:06.94elapsed 79%CPU
// flush   10000, sync=NORMAL -> 51.37user 2.03system 1:31.63elapsed 58%CPU
// flush    1000, sync=NORMAL -> 52.16user 2.70system 4:38.58elapsed 19%CPU
 
//         MIDAS includes             //
 
#include "midas.h"
#include "history.h"
 
//           helper stuff             //
 
#define FREE(x) { if (x) free(x); (x) = NULL; }
 
static char* skip_spaces(char* s)
{
   while (*s) {
      if (!isspace(*s))
         break;
      s++;
   }
   return s;
}
 
static std::string TimeToString(time_t t)
{
   const char* sign = "";
 
   if (t == 0)
      return "0";
 
   time_t tt = t;
 
   if (t < 0) {
      sign = "-";
      tt = -t;
   }
 
   assert(tt > 0);
 
   std::string v;
   while (tt) {
      char c = '0' + (char)(tt%10);
      tt /= 10;
      v = c + v;
   }
 
   v = sign + v;
 
   //printf("time %.0f -> %s\n", (double)t, v.c_str());
 
   return v;
}
 
static std::string SmallIntToString(int i)
{
   //int ii = i;
 
   if (i == 0)
      return "0";
 
   assert(i > 0);
 
   std::string v;
   while (i) {
      char c = '0' + (char)(i%10);
      i /= 10;
      v = c + v;
   }
 
   //printf("SmallIntToString: %d -> %s\n", ii, v.c_str());
 
   return v;
}
 
static bool MatchEventName(const char* event_name, const char* var_event_name)
{
   // new-style event name: "equipment_name/variable_name:tag_name"
   // old-style event name: "equipment_name:tag_name" ("variable_name" is missing)
   bool newStyleEventName = (strchr(var_event_name, '/')!=NULL);
 
   //printf("looking for event_name [%s], try table [%s] event name [%s], new style [%d]\n", var_event_name, table_name, event_name, newStyleEventName);
 
   if (strcasecmp(event_name, var_event_name) == 0) {
      return true;
   } else if (newStyleEventName) {
      return false;
   } else { // for old style names, need more parsing
      bool match = false;
 
      const char* s = event_name;
      for (int j=0; s[j]; j++) {
 
         if ((var_event_name[j]==0) && (s[j]=='/')) {
            match = true;
            break;
         }
 
         if ((var_event_name[j]==0) && (s[j]=='_')) {
            match = true;
            break;
         }
 
         if (var_event_name[j]==0) {
            match = false;
            break;
         }
 
         if (tolower(var_event_name[j]) != tolower(s[j])) { // does not work for UTF-8 Unicode
            match = false;
            break;
         }
      }
 
      return match;
   }
}
 
static bool MatchTagName(const char* tag_name, int n_data, const char* var_tag_name, const int var_tag_index)
{
   char alt_tag_name[1024]; // maybe this is an array without "Names"?
   sprintf(alt_tag_name, "%s[%d]", var_tag_name, var_tag_index);
 
   //printf("  looking for tag [%s] alt [%s], try column name [%s]\n", var_tag_name, alt_tag_name, tag_name);
 
   if (strcasecmp(tag_name, var_tag_name) == 0)
      if (var_tag_index >= 0 && var_tag_index < n_data)
         return true;
 
   if (strcasecmp(tag_name, alt_tag_name) == 0)
      return true;
 
   return false;
}
 
static void PrintTags(int ntags, const TAG tags[])
{
   for (int i=0; i<ntags; i++)
      printf("tag %d: %s %s[%d]\n", i, rpc_tid_name(tags[i].type), tags[i].name, tags[i].n_data);
}
 
// convert MIDAS event name to something acceptable as an SQL identifier - table name, column name, etc
 
static std::string MidasNameToSqlName(const char* s)
{
   std::string out;
 
   for (int i=0; s[i]!=0; i++) {
      char c = s[i];
      if (isalpha(c) || isdigit(c))
         out += tolower(c); // does not work for UTF-8 Unicode
      else
         out += '_';
   }
 
   return out;
}
 
// convert MIDAS event name to something acceptable as a file name
 
static std::string MidasNameToFileName(const char* s)
{
   std::string out;
 
   for (int i=0; s[i]!=0; i++) {
      char c = s[i];
      if (isalpha(c) || isdigit(c))
         out += tolower(c); // does not work for UTF-8 Unicode
      else
         out += '_';
   }
 
   return out;
}
 
// compare event names
 
static int event_name_cmp(const std::string& e1, const char* e2)
{
   return strcasecmp(e1.c_str(), e2);
}
 
// compare variable names
 
static int var_name_cmp(const std::string& v1, const char* v2)
{
   return strcasecmp(v1.c_str(), v2);
}
 
//         SQL data types             //
 
#ifdef HAVE_SQLITE
static const char *sql_type_sqlite[TID_LAST] = {
   "xxxINVALIDxxxNULL", // TID_NULL
   "INTEGER",           // TID_UINT8
   "INTEGER",           // TID_INT8
   "TEXT",              // TID_CHAR
   "INTEGER",           // TID_UINT16
   "INTEGER",           // TID_INT16
   "INTEGER",           // TID_UINT32
   "INTEGER",           // TID_INT32
   "INTEGER",           // TID_BOOL
   "REAL",              // TID_FLOAT
   "REAL",              // TID_DOUBLE
   "INTEGER",           // TID_BITFIELD
   "TEXT",              // TID_STRING
   "xxxINVALIDxxxARRAY",
   "xxxINVALIDxxxSTRUCT",
   "xxxINVALIDxxxKEY",
   "xxxINVALIDxxxLINK"
};
#endif
 
#ifdef HAVE_PGSQL
static const char *sql_type_pgsql[TID_LAST] = {
   "xxxINVALIDxxxNULL", // TID_NULL
   "smallint",  // TID_BYTE
   "smallint",  // TID_SBYTE
   "char(1)",   // TID_CHAR
   "integer",   // TID_WORD
   "smallint",  // TID_SHORT
   "bigint",    // TID_DWORD
   "integer",   // TID_INT
   "smallint",  // TID_BOOL
   "real",      // TID_FLOAT
   "double precision", // TID_DOUBLE
   "bigint",    // TID_BITFIELD
   "text",      // TID_STRING
   "xxxINVALIDxxxARRAY",
   "xxxINVALIDxxxSTRUCT",
   "xxxINVALIDxxxKEY",
   "xxxINVALIDxxxLINK"
};
#endif
 
#ifdef HAVE_MYSQL
static const char *sql_type_mysql[TID_LAST] = {
   "xxxINVALIDxxxNULL", // TID_NULL
   "tinyint unsigned",  // TID_BYTE
   "tinyint",           // TID_SBYTE
   "char",              // TID_CHAR
   "smallint unsigned", // TID_WORD
   "smallint",          // TID_SHORT
   "integer unsigned",  // TID_DWORD
   "integer",           // TID_INT
   "tinyint",           // TID_BOOL
   "float",             // TID_FLOAT
   "double",            // TID_DOUBLE
   "integer unsigned",  // TID_BITFIELD
   "VARCHAR",           // TID_STRING
   "xxxINVALIDxxxARRAY",
   "xxxINVALIDxxxSTRUCT",
   "xxxINVALIDxxxKEY",
   "xxxINVALIDxxxLINK"
};
#endif
 
void DoctorPgsqlColumnType(std::string* col_type, const char* index_type)
{
   if (*col_type == index_type)
      return;
 
   if (*col_type == "bigint" && strcmp(index_type, "int8")==0) {
      *col_type = index_type;
      return;
   }
 
   if (*col_type == "integer" && strcmp(index_type, "int4")==0) {
      *col_type = index_type;
      return;
   }
 
   if (*col_type == "smallint" && strcmp(index_type, "int2")==0) {
      *col_type = index_type;
      return;
   }
 
   cm_msg(MERROR, "SqlHistory", "Cannot use this SQL database, incompatible column names: created column type [%s] is reported with column type [%s]", index_type, col_type->c_str());
   cm_msg_flush_buffer();
   abort();
}
 
void DoctorSqlColumnType(std::string* col_type, const char* index_type)
{
   if (*col_type == index_type)
      return;
 
   if (*col_type == "int(10) unsigned" && strcmp(index_type, "integer unsigned")==0) {
      *col_type = index_type;
      return;
   }
 
   if (*col_type == "int(11)" && strcmp(index_type, "integer")==0) {
      *col_type = index_type;
      return;
   }
 
   if (*col_type == "integer" && strcmp(index_type, "int(11)")==0) {
      *col_type = index_type;
      return;
   }
 
   // MYSQL 8.0.23
 
   if (*col_type == "int" && strcmp(index_type, "integer")==0) {
      *col_type = index_type;
      return;
   }
 
   if (*col_type == "int unsigned" && strcmp(index_type, "integer unsigned")==0) {
      *col_type = index_type;
      return;
   }
 
   cm_msg(MERROR, "SqlHistory", "Cannot use this SQL database, incompatible column names: created column type [%s] is reported with column type [%s]", index_type, col_type->c_str());
   cm_msg_flush_buffer();
   abort();
}
 
#if 0
static int sql2midasType_mysql(const char* name)
{
   for (int tid=0; tid<TID_LAST; tid++)
      if (strcasecmp(name, sql_type_mysql[tid])==0)
         return tid;
   // FIXME!
   printf("sql2midasType: Cannot convert SQL data type \'%s\' to a MIDAS data type!\n", name);
   return 0;
}
#endif
 
#if 0
static int sql2midasType_sqlite(const char* name)
{
   if (strcmp(name, "INTEGER") == 0)
      return TID_INT;
   if (strcmp(name, "REAL") == 0)
      return TID_DOUBLE;
   if (strcmp(name, "TEXT") == 0)
      return TID_STRING;
   // FIXME!
   printf("sql2midasType: Cannot convert SQL data type \'%s\' to a MIDAS data type!\n", name);
   return 0;
}
#endif
 
//        Schema base classes         //
 
struct HsSchemaEntry {
   std::string tag_name; // tag name from MIDAS
   std::string tag_type; // tag type from MIDAS
   std::string name;     // entry name, same as tag_name except when read from SQL history when it could be the SQL column name
   int type    = 0; // MIDAS data type TID_xxx
   int n_data  = 0; // MIDAS array size
   int n_bytes = 0; // n_data * size of MIDAS data type (only used by HsFileSchema?)
};
 
class HsSchema
{
public:
 
   // event schema definitions
   std::string fEventName;
   time_t      fTimeFrom = 0;
   time_t      fTimeTo = 0;
   std::vector<HsSchemaEntry> fVariables;
   std::vector<int> fOffsets;
   size_t      fNumBytes = 0;
 
   // run time data used by hs_write_event()
   int fCountWriteUndersize = 0;
   int fCountWriteOversize = 0;
   size_t fWriteMaxSize = 0;
   size_t fWriteMinSize = 0;
 
   // schema disabled by write error
   bool fDisabled = true;
 
public:
 
   HsSchema() // ctor
   {
      // empty
   }
 
   virtual void remove_inactive_columns() = 0; // used by SQL schemas
   virtual void print(bool print_tags = true) const;
   virtual ~HsSchema(); // dtor
   virtual int flush_buffers() = 0;
   virtual int close() = 0;
   virtual int write_event(const time_t t, const char* data, const size_t data_size) = 0;
   virtual int match_event_var(const char* event_name, const char* var_name, const int var_index);
   virtual int read_last_written(const time_t timestamp,
                                 const int debug,
                                 time_t* last_written) = 0;
   virtual int read_data(const time_t start_time,
                         const time_t end_time,
                         const int num_var, const std::vector<int>& var_schema_index, const int var_index[],
                         const int debug,
                         std::vector<time_t>& last_time,
                         MidasHistoryBufferInterface* buffer[]) = 0;
};
 
class HsSchemaVector
{
protected:
   std::vector<HsSchema*> fData;
 
public:
   ~HsSchemaVector() { // dtor
      clear();
   }
 
   HsSchema* operator[](int index) const {
      return fData[index];
   }
 
   size_t size() const {
      return fData.size();
   }
 
   void add(HsSchema* s);
 
   void clear() {
      for (size_t i=0; i<fData.size(); i++)
         if (fData[i]) {
            delete fData[i];
            fData[i] = NULL;
         }
      fData.clear();
   }
 
   void print(bool print_tags = true) const {
      for (size_t i=0; i<fData.size(); i++)
         fData[i]->print(print_tags);
   }
 
   HsSchema* find_event(const char* event_name, const time_t timestamp, int debug = 0);
};
 
//        Base class functions            //
 
HsSchema::~HsSchema() // dtor
{
   //printf("HsSchema::dtor %p!\n", this);
   // only report if undersize/oversize happens more than once -
   // the first occurence is already reported by hs_write_event()
   if (fCountWriteUndersize > 1) {
      cm_msg(MERROR, "hs_write_event", "Event \'%s\' data size mismatch count: %d, expected %zu bytes, hs_write_event() called with as few as %zu bytes", fEventName.c_str(), fCountWriteUndersize, fNumBytes, fWriteMinSize);
   }
 
   if (fCountWriteOversize > 1) {
      cm_msg(MERROR, "hs_write_event", "Event \'%s\' data size mismatch count: %d, expected %zu bytes, hs_write_event() called with as many as %zu bytes", fEventName.c_str(), fCountWriteOversize, fNumBytes, fWriteMaxSize);
   }
};
 
void HsSchemaVector::add(HsSchema* s)
{
   // schema list "data" is sorted by decreasing "fTimeFrom", newest schema first
 
   //printf("add: %s..%s %s\n", TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), s->fEventName.c_str());
 
   bool added = false;
 
   for (auto it = fData.begin(); it != fData.end(); it++) {
      if (event_name_cmp((*it)->fEventName, s->fEventName.c_str())==0) {
         if (s->fTimeFrom == (*it)->fTimeFrom) {
            // duplicate schema, keep the last one added (for file schema it is the newer file)
            s->fTimeTo = (*it)->fTimeTo;
            delete (*it);
            (*it) = s;
            return;
         }
      }
 
      if (s->fTimeFrom > (*it)->fTimeFrom) {
         fData.insert(it, s);
         added = true;
         break;
      }
   }
 
   if (!added) {
      fData.push_back(s);
   }
 
   //time_t oldest_time_from = fData.back()->fTimeFrom;
 
   time_t time_to = 0;
 
   for (auto it = fData.begin(); it != fData.end(); it++) {
      if (event_name_cmp((*it)->fEventName, s->fEventName.c_str())==0) {
         (*it)->fTimeTo = time_to;
         time_to = (*it)->fTimeFrom;
 
         //printf("vvv: %s..%s %s\n", TimeToString((*it)->fTimeFrom-oldest_time_from).c_str(), TimeToString((*it)->fTimeTo-oldest_time_from).c_str(), (*it)->fEventName.c_str());
      }
   }
}
 
HsSchema* HsSchemaVector::find_event(const char* event_name, time_t t, int debug)
{
   HsSchema* ss = NULL;
 
   if (debug) {
      printf("find_event: All schema for event %s: (total %zu)\n", event_name, fData.size());
      int found = 0;
      for (size_t i=0; i<fData.size(); i++) {
         HsSchema* s = fData[i];
         printf("find_event: schema %zu name [%s]\n", i, s->fEventName.c_str());
         if (event_name)
            if (event_name_cmp(s->fEventName, event_name)!=0)
               continue;
         s->print();
         found++;
      }
      printf("find_event: Found %d schemas for event %s\n", found, event_name);
 
      //if (found == 0)
      //   abort();
   }
 
   for (size_t i=0; i<fData.size(); i++) {
      HsSchema* s = fData[i];
 
      // wrong event
      if (event_name)
         if (event_name_cmp(s->fEventName, event_name)!=0)
            continue;
 
      // schema is from after the time we are looking for
      if (s->fTimeFrom > t)
         continue;
 
      if (!ss)
         ss = s;
 
      // remember the newest schema
      if (s->fTimeFrom > ss->fTimeFrom)
         ss = s;
   }
 
   // try to find
   for (size_t i=0; i<fData.size(); i++) {
      HsSchema* s = fData[i];
 
      // wrong event
      if (event_name)
         if (event_name_cmp(s->fEventName, event_name)!=0)
            continue;
 
      // schema is from after the time we are looking for
      if (s->fTimeFrom > t)
         continue;
 
      if (!ss)
         ss = s;
 
      // remember the newest schema
      if (s->fTimeFrom > ss->fTimeFrom)
         ss = s;
   }
 
   if (debug) {
      if (ss) {
         printf("find_event: for time %s, returning:\n", TimeToString(t).c_str());
         ss->print();
      } else {
         printf("find_event: for time %s, nothing found:\n", TimeToString(t).c_str());
      }
   }
 
   return ss;
}
 
//        Sql interface class             //
 
class SqlBase
{
public:
   int  fDebug = 0;
   bool fIsConnected = false;
   bool fTransactionPerTable = true;
 
   SqlBase() {  // ctor
   };
 
   virtual ~SqlBase() { // dtor
      // confirm that the destructor of the concrete class
      // disconnected the database
      assert(!fIsConnected);
      fDebug = 0;
      fIsConnected = false;
   }
 
   virtual int Connect(const char* path) = 0;
   virtual int Disconnect() = 0;
   virtual bool IsConnected() = 0;
 
   virtual int ListTables(std::vector<std::string> *plist) = 0;
   virtual int ListColumns(const char* table_name, std::vector<std::string> *plist) = 0;
 
   // sql commands
   virtual int Exec(const char* table_name, const char* sql) = 0;
   virtual int ExecDisconnected(const char* table_name, const char* sql) = 0;
 
   // queries
   virtual int Prepare(const char* table_name, const char* sql) = 0;
   virtual int Step() = 0;
   virtual const char* GetText(int column) = 0;
   virtual time_t      GetTime(int column) = 0;
   virtual double      GetDouble(int column) = 0;
   virtual int Finalize() = 0;
 
   // transactions
   virtual int OpenTransaction(const char* table_name) = 0;
   virtual int CommitTransaction(const char* table_name) = 0;
   virtual int RollbackTransaction(const char* table_name) = 0;
 
   // data types
   virtual const char* ColumnType(int midas_tid) = 0;
   virtual bool TypesCompatible(int midas_tid, const char* sql_type) = 0;
 
   // string quoting
   virtual std::string QuoteString(const char* s) = 0; // quote text string
   virtual std::string QuoteId(const char* s) = 0; // quote identifier, such as table or column name
};
 
//        Schema concrete classes         //
 
class HsSqlSchema : public HsSchema
{
public:
 
   SqlBase* fSql = NULL;
   std::string fTableName;
   std::vector<std::string> fColumnNames;
   std::vector<std::string> fColumnTypes;
   std::vector<bool>        fColumnInactive;
 
   // Cache: SQL column name -> index into fColumnNames/fColumnTypes/fColumnInactive/fVariables/fOffsets.
   // Rebuilt lazily when it is out of sync with fColumnNames (see find_column_index()).
   std::unordered_map<std::string, size_t> fColumnIndexCache;
 
public:
 
   HsSqlSchema() // ctor
   {
      // empty
   }
 
   ~HsSqlSchema() // dtor
   {
      assert(get_transaction_count() == 0);
   }
 
   void remove_inactive_columns();
   int find_column_index(const std::string& column_name);
   void print(bool print_tags = true) const;
   int get_transaction_count();
   void reset_transaction_count();
   void increment_transaction_count();
   int close_transaction();
   int flush_buffers() { return close_transaction(); }
   int close() { return close_transaction(); }
   int write_event(const time_t t, const char* data, const size_t data_size);
   int match_event_var(const char* event_name, const char* var_name, const int var_index);
   int read_last_written(const time_t timestamp,
                         const int debug,
                         time_t* last_written);
   int read_data(const time_t start_time,
                 const time_t end_time,
                 const int num_var, const std::vector<int>& var_schema_index, const int var_index[],
                 const int debug,
                 std::vector<time_t>& last_time,
                 MidasHistoryBufferInterface* buffer[]);
 
private:
   // Sqlite uses a transaction per table; MySQL uses a single transaction for all tables.
   // But to support future "single transaction" DBs more easily (e.g. if user wants to
   // log to both Postgres and MySQL in future), we keep track of the transaction count
   // per SQL engine.
   int fTableTransactionCount = 0;
   static std::map<SqlBase*, int> gfTransactionCount;
};
 
std::map<SqlBase*, int> HsSqlSchema::gfTransactionCount;
 
int HsSqlSchema::find_column_index(const std::string& column_name)
{
   // Rebuild the cache if it is out of sync with fColumnNames. Columns are only
   // ever appended (read_column_names) or removed in bulk with a resize
   // (remove_inactive_columns), so a size mismatch reliably detects staleness.
   if (fColumnIndexCache.size() != fColumnNames.size()) {
      fColumnIndexCache.clear();
      for (size_t i=0; i<fColumnNames.size(); i++)
         fColumnIndexCache[fColumnNames[i]] = i;
   }
 
   auto it = fColumnIndexCache.find(column_name);
   if (it == fColumnIndexCache.end())
      return -1;
   return (int)it->second;
}
 
class HsFileSchema : public HsSchema
{
public:
 
   std::string fFileName;
   size_t  fRecordSize = 0;
   off64_t fDataOffset = 0;
   size_t  fLastSize   = 0;
   int     fWriterFd   = -1;
   size_t  fRecordBufferSize = 0;
   char*   fRecordBuffer = NULL;
 
public:
   off64_t fFileSizeInitial = 0; // initial file size
   off64_t fFileSize = 0; // file size including any new data we wrote
 
public:
 
   HsFileSchema() // ctor
   {
      // empty
   }
 
   ~HsFileSchema() // dtor
   {
      //printf("HsFileSchema::dtor %p!\n", this);
      close();
      fRecordSize = 0;
      fDataOffset = 0;
      fLastSize = 0;
      fWriterFd = -1;
      if (fRecordBuffer) {
         free(fRecordBuffer);
         fRecordBuffer = NULL;
      }
      fRecordBufferSize = 0;
   }
 
   void remove_inactive_columns() { /* empty */ };
   void print(bool print_tags = true) const;
   int flush_buffers() { return HS_SUCCESS; };
   int close();
   int write_event(const time_t t, const char* data, const size_t data_size);
   int read_last_written(const time_t timestamp,
                         const int debug,
                         time_t* last_written);
   int read_data(const time_t start_time,
                 const time_t end_time,
                 const int num_var, const std::vector<int>& var_schema_index, const int var_index[],
                 const int debug,
                 std::vector<time_t>& last_time,
                 MidasHistoryBufferInterface* buffer[]);
};
 
//        Print functions                 //
 
void HsSchema::print(bool print_tags) const
{
   size_t nv = this->fVariables.size();
   printf("event [%s], time %s..%s, %zu variables, %zu bytes\n", this->fEventName.c_str(), TimeToString(this->fTimeFrom).c_str(), TimeToString(this->fTimeTo).c_str(), nv, fNumBytes);
   if (print_tags)
      for (size_t j=0; j<nv; j++)
         printf("  %zu: name [%s], type [%s] tid %d, n_data %d, n_bytes %d, offset %d\n", j, this->fVariables[j].name.c_str(), rpc_tid_name(this->fVariables[j].type), this->fVariables[j].type, this->fVariables[j].n_data, this->fVariables[j].n_bytes, this->fOffsets[j]);
};
 
void HsSqlSchema::print(bool print_tags) const
{
   size_t nv = this->fVariables.size();
   printf("event [%s], sql_table [%s], time %s..%s, %zu variables, %zu bytes\n", this->fEventName.c_str(), this->fTableName.c_str(), TimeToString(this->fTimeFrom).c_str(), TimeToString(this->fTimeTo).c_str(), nv, fNumBytes);
   if (print_tags) {
      for (size_t j=0; j<nv; j++) {
         printf("  %zu: name [%s], type [%s] tid %d, n_data %d, n_bytes %d", j, this->fVariables[j].name.c_str(), rpc_tid_name(this->fVariables[j].type), this->fVariables[j].type, this->fVariables[j].n_data, this->fVariables[j].n_bytes);
         printf(", sql_column [%s], sql_type [%s], offset %d", this->fColumnNames[j].c_str(), this->fColumnTypes[j].c_str(), this->fOffsets[j]);
         printf(", inactive %d", (int)this->fColumnInactive[j]);
         printf("\n");
      }
   }
}
 
void HsFileSchema::print(bool print_tags) const
{
   size_t nv = this->fVariables.size();
   printf("event [%s], file_name [%s], time %s..%s, %zu variables, %zu bytes, dat_offset %jd, record_size %zu\n", this->fEventName.c_str(), this->fFileName.c_str(), TimeToString(this->fTimeFrom).c_str(), TimeToString(this->fTimeTo).c_str(), nv, fNumBytes, (intmax_t)fDataOffset, fRecordSize);
   if (print_tags) {
      for (size_t j=0; j<nv; j++)
         printf("  %zu: name [%s], type [%s] tid %d, n_data %d, n_bytes %d, offset %d\n", j, this->fVariables[j].name.c_str(), rpc_tid_name(this->fVariables[j].type), this->fVariables[j].type, this->fVariables[j].n_data, this->fVariables[j].n_bytes, this->fOffsets[j]);
   }
}
 
//        File functions                  //
 
#ifdef HAVE_MYSQL
 
//     MYSQL/MariaDB database access  //
 
//#warning !!!HAVE_MYSQL!!!
 
//#include <my_global.h> // my_global.h removed MySQL 8.0, MariaDB 10.2. K.O.
#include <mysql.h>
 
class Mysql: public SqlBase
{
public:
   std::string fConnectString;
   MYSQL* fMysql = NULL;
 
   // query results
   MYSQL_RES* fResult = NULL;
   MYSQL_ROW  fRow = NULL;
   int fNumFields = 0;
 
   // disconnected operation
   size_t fMaxDisconnected = 0;
   std::list<std::string> fDisconnectedBuffer;
   time_t fNextReconnect = 0;
   int fNextReconnectDelaySec = 0;
   int fDisconnectedLost = 0;
 
   Mysql(); // ctor
   ~Mysql(); // dtor
 
   int Connect(const char* path);
   int Disconnect();
   bool IsConnected();
 
   int ConnectTable(const char* table_name);
 
   int ListTables(std::vector<std::string> *plist);
   int ListColumns(const char* table_name, std::vector<std::string> *plist);
 
   int Exec(const char* table_name, const char* sql);
   int ExecDisconnected(const char* table_name, const char* sql);
 
   int Prepare(const char* table_name, const char* sql);
   int Step();
   const char* GetText(int column);
   time_t      GetTime(int column);
   double      GetDouble(int column);
   int Finalize();
 
   int OpenTransaction(const char* table_name);
   int CommitTransaction(const char* table_name);
   int RollbackTransaction(const char* table_name);
 
   const char* ColumnType(int midas_tid);
   bool TypesCompatible(int midas_tid, const char* sql_type);
 
   std::string QuoteId(const char* s);
   std::string QuoteString(const char* s);
};
 
Mysql::Mysql() // ctor
{
   fMysql = NULL;
   fResult = NULL;
   fRow = NULL;
   fNumFields = 0;
   fMaxDisconnected = 1000;
   fNextReconnect = 0;
   fNextReconnectDelaySec = 0;
   fDisconnectedLost = 0;
   fTransactionPerTable = false;
}
 
Mysql::~Mysql() // dtor
{
   Disconnect();
   fMysql = NULL;
   fResult = NULL;
   fRow = NULL;
   fNumFields = 0;
   if (fDisconnectedBuffer.size() > 0) {
      cm_msg(MINFO, "Mysql::~Mysql", "Lost %zu history entries accumulated while disconnected from the database", fDisconnectedBuffer.size());
      cm_msg_flush_buffer();
   }
}
 
int Mysql::Connect(const char* connect_string)
{
   if (fIsConnected)
      Disconnect();
 
   fConnectString = connect_string;
 
   if (fDebug) {
      cm_msg(MINFO, "Mysql::Connect", "Connecting to Mysql database specified by \'%s\'", connect_string);
      cm_msg_flush_buffer();
   }
 
   std::string host_name;
   std::string user_name;
   std::string user_password;
   std::string db_name;
   int tcp_port = 0;
   std::string unix_socket;
   std::string buffer;
 
   FILE* fp = fopen(connect_string, "r");
   if (!fp) {
      cm_msg(MERROR, "Mysql::Connect", "Cannot read MYSQL connection parameters from \'%s\', fopen() error %d (%s)", connect_string, errno, strerror(errno));
      return DB_FILE_ERROR;
   }
 
   while (1) {
      char buf[256];
      char* s = fgets(buf, sizeof(buf), fp);
      if (!s)
         break; // EOF
 
      char*ss;
      // kill trailing \n and \r
      ss = strchr(s, '\n');
      if (ss) *ss = 0;
      ss = strchr(s, '\r');
      if (ss) *ss = 0;
 
      //printf("line [%s]\n", s);
 
      if (strncasecmp(s, "server=", 7)==0)
         host_name = skip_spaces(s + 7);
      if (strncasecmp(s, "port=", 5)==0)
         tcp_port = atoi(skip_spaces(s + 5));
      if (strncasecmp(s, "database=", 9)==0)
         db_name = skip_spaces(s + 9);
      if (strncasecmp(s, "socket=", 7)==0)
         unix_socket = skip_spaces(s + 7);
      if (strncasecmp(s, "user=", 5)==0)
         user_name = skip_spaces(s + 5);
      if (strncasecmp(s, "password=", 9)==0)
         user_password = skip_spaces(s + 9);
      if (strncasecmp(s, "buffer=", 7)==0)
         buffer = skip_spaces(s + 7);
   }
 
   fclose(fp);
 
   int buffer_int = atoi(buffer.c_str());
 
   if (buffer_int > 0 && buffer_int < 1000000)
      fMaxDisconnected = buffer_int;
 
   if (fDebug)
      printf("Mysql::Connect: connecting to server [%s] port %d, unix socket [%s], database [%s], user [%s], password [%s], buffer [%zu]\n", host_name.c_str(), tcp_port, unix_socket.c_str(), db_name.c_str(), user_name.c_str(), user_password.c_str(), fMaxDisconnected);
 
   if (!fMysql) {
      fMysql = mysql_init(NULL);
      if (!fMysql) {
         return DB_FILE_ERROR;
      }
   }
 
   int client_flag = 0|CLIENT_IGNORE_SIGPIPE;
 
   if (mysql_real_connect(fMysql, host_name.c_str(), user_name.c_str(), user_password.c_str(), db_name.c_str(), tcp_port, unix_socket.c_str(), client_flag) == NULL) {
      cm_msg(MERROR, "Mysql::Connect", "mysql_real_connect() to host [%s], port %d, unix socket [%s], database [%s], user [%s], password [%s]: error %d (%s)", host_name.c_str(), tcp_port, unix_socket.c_str(), db_name.c_str(), user_name.c_str(), "xxx", mysql_errno(fMysql), mysql_error(fMysql));
      Disconnect();
      return DB_FILE_ERROR;
   }
 
   int status;
 
   // FIXME:
   //my_bool reconnect = 0;
   //mysql_options(&mysql, MYSQL_OPT_RECONNECT, &reconnect);
 
   status = Exec("(notable)", "SET SESSION sql_mode='ANSI'");
   if (status != DB_SUCCESS) {
      cm_msg(MERROR, "Mysql::Connect", "Cannot set ANSI mode, nothing will work");
      Disconnect();
      return DB_FILE_ERROR;
   }
 
   if (fDebug) {
      cm_msg(MINFO, "Mysql::Connect", "Connected to a MySQL database on host [%s], port %d, unix socket [%s], database [%s], user [%s], password [%s], buffer %zu", host_name.c_str(), tcp_port, unix_socket.c_str(), db_name.c_str(), user_name.c_str(), "xxx", fMaxDisconnected);
      cm_msg_flush_buffer();
   }
 
   fIsConnected = true;
 
   int count = 0;
   while (fDisconnectedBuffer.size() > 0) {
      status = Exec("(flush)", fDisconnectedBuffer.front().c_str());
      if (status != DB_SUCCESS) {
         return status;
      }
      fDisconnectedBuffer.pop_front();
      count++;
   }
 
   if (count > 0) {
      cm_msg(MINFO, "Mysql::Connect", "Saved %d, lost %d history events accumulated while disconnected from the database", count, fDisconnectedLost);
      cm_msg_flush_buffer();
   }
 
   assert(fDisconnectedBuffer.size() == 0);
   fDisconnectedLost = 0;
 
   return DB_SUCCESS;
}
 
int Mysql::Disconnect()
{
   if (fRow) {
      // FIXME: mysql_free_result(fResult);
   }
 
   if (fResult)
      mysql_free_result(fResult);
 
   if (fMysql)
      mysql_close(fMysql);
 
   fMysql = NULL;
   fResult = NULL;
   fRow = NULL;
 
   fIsConnected = false;
   return DB_SUCCESS;
}
 
bool Mysql::IsConnected()
{
   return fIsConnected;
}
 
int Mysql::OpenTransaction(const char* table_name)
{
   return Exec(table_name, "START TRANSACTION");
   return DB_SUCCESS;
}
 
int Mysql::CommitTransaction(const char* table_name)
{
   Exec(table_name, "COMMIT");
   return DB_SUCCESS;
}
 
int Mysql::RollbackTransaction(const char* table_name)
{
   Exec(table_name, "ROLLBACK");
   return DB_SUCCESS;
}
 
int Mysql::ListTables(std::vector<std::string> *plist)
{
   if (!fIsConnected)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Mysql::ListTables!\n");
 
   int status;
 
   fResult = mysql_list_tables(fMysql, NULL);
 
   if (fResult == NULL) {
      cm_msg(MERROR, "Mysql::ListTables", "mysql_list_tables() error %d (%s)", mysql_errno(fMysql), mysql_error(fMysql));
      return DB_FILE_ERROR;
   }
 
   fNumFields = mysql_num_fields(fResult);
 
   while (1) {
      status = Step();
      if (status != DB_SUCCESS)
         break;
      std::string tn = GetText(0);
      plist->push_back(tn);
   };
 
   status = Finalize();
 
   return DB_SUCCESS;
}
 
int Mysql::ListColumns(const char* table_name, std::vector<std::string> *plist)
{
   if (!fIsConnected)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Mysql::ListColumns for table \'%s\'\n", table_name);
 
   int status;
 
   std::string cmd;
   cmd += "SHOW COLUMNS FROM ";
   cmd += QuoteId(table_name);
   cmd += ";";
 
   status = Prepare(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return status;
 
   fNumFields = mysql_num_fields(fResult);
 
   while (1) {
      status = Step();
      if (status != DB_SUCCESS)
         break;
      std::string cn = GetText(0);
      std::string ct = GetText(1);
      plist->push_back(cn);
      plist->push_back(ct);
      //printf("cn [%s]\n", cn.c_str());
      //for (int i=0; i<fNumFields; i++)
      //printf(" field[%d]: [%s]\n", i, GetText(i));
   };
 
   status = Finalize();
 
   return DB_SUCCESS;
}
 
int Mysql::Exec(const char* table_name, const char* sql)
{
   if (fDebug)
      printf("Mysql::Exec(%s, %s)\n", table_name, sql);
 
   // FIXME: match Sqlite::Exec() return values:
   // return values:
   // DB_SUCCESS
   // DB_FILE_ERROR: not connected
   // DB_KEY_EXIST: "table already exists"
 
   if (!fMysql)
      return DB_FILE_ERROR;
 
   assert(fMysql);
   assert(fResult == NULL); // there should be no unfinalized queries
   assert(fRow == NULL);
 
   if (mysql_query(fMysql, sql)) {
      if (mysql_errno(fMysql) == 1050) { // "Table already exists"
         return DB_KEY_EXIST;
      }         
      if (mysql_errno(fMysql) == 1146) { // "Table does not exist"
         return DB_FILE_ERROR;
      }         
      cm_msg(MERROR, "Mysql::Exec", "mysql_query(%s) error %d (%s)", sql, mysql_errno(fMysql), mysql_error(fMysql));
      if (mysql_errno(fMysql) == 1060) // "Duplicate column name"
         return DB_KEY_EXIST;
      if (mysql_errno(fMysql) == 2006) { // "MySQL server has gone away"
         Disconnect();
         return ExecDisconnected(table_name, sql);
      }
      return DB_FILE_ERROR;
   }
 
   return DB_SUCCESS;
}
 
int Mysql::ExecDisconnected(const char* table_name, const char* sql)
{
   if (fDebug)
      printf("Mysql::ExecDisconnected(%s, %s)\n", table_name, sql);
 
   if (fDisconnectedBuffer.size() < fMaxDisconnected) {
      fDisconnectedBuffer.push_back(sql);
      if (fDisconnectedBuffer.size() >= fMaxDisconnected) {
         cm_msg(MERROR, "Mysql::ExecDisconnected", "Error: Disconnected database buffer overflow, size %zu, subsequent events are lost", fDisconnectedBuffer.size());
      }
   } else {
      fDisconnectedLost++;
   }
 
   time_t now = time(NULL);
 
   if (fNextReconnect == 0 || now >= fNextReconnect) {
      int status = Connect(fConnectString.c_str());
      if (status == DB_SUCCESS) {
         fNextReconnect = 0;
         fNextReconnectDelaySec = 0;
      } else {
         if (fNextReconnectDelaySec == 0) {
            fNextReconnectDelaySec = 5;
         } else if (fNextReconnectDelaySec < 10*60) {
            fNextReconnectDelaySec *= 2;
         }
         if (fDebug) {
            cm_msg(MINFO, "Mysql::ExecDisconnected", "Next reconnect attempt in %d sec, history events buffered %zu, lost %d", fNextReconnectDelaySec, fDisconnectedBuffer.size(), fDisconnectedLost);
            cm_msg_flush_buffer();
         }
         fNextReconnect = now + fNextReconnectDelaySec;
      }
   }
   
   return DB_SUCCESS;
}
 
int Mysql::Prepare(const char* table_name, const char* sql)
{
   if (fDebug)
      printf("Mysql::Prepare(%s, %s)\n", table_name, sql);
 
   if (!fMysql)
      return DB_FILE_ERROR;
 
   assert(fMysql);
   assert(fResult == NULL); // there should be no unfinalized queries
   assert(fRow == NULL);
 
   //   if (mysql_query(fMysql, sql)) {
   //   cm_msg(MERROR, "Mysql::Prepare", "mysql_query(%s) error %d (%s)", sql, mysql_errno(fMysql), mysql_error(fMysql));
   //   return DB_FILE_ERROR;
   //}
 
   // Check if the connection to MySQL timed out; fix from B. Smith
   int status = mysql_query(fMysql, sql);
   if (status) {
      if (mysql_errno(fMysql) == 2006 || mysql_errno(fMysql) == 2013) {
         // "MySQL server has gone away" or "Lost connection to MySQL server during query"
         status = Connect(fConnectString.c_str());
         if (status == DB_SUCCESS) {
            // Retry after reconnecting
            status = mysql_query(fMysql, sql);
         } else {
            cm_msg(MERROR, "Mysql::Prepare", "mysql_query(%s) - MySQL server has gone away, and couldn't reconnect - %d", sql, status);
            return DB_FILE_ERROR;
         }
      }
      if (status) {
         cm_msg(MERROR, "Mysql::Prepare", "mysql_query(%s) error %d (%s)", sql, mysql_errno(fMysql), mysql_error(fMysql));
         return DB_FILE_ERROR;
      }
      cm_msg(MINFO, "Mysql::Prepare", "Reconnected to MySQL after long inactivity.");
   }
 
   fResult = mysql_store_result(fMysql);
   //fResult = mysql_use_result(fMysql); // cannot use this because it blocks writing into table
 
   if (!fResult) {
      cm_msg(MERROR, "Mysql::Prepare", "mysql_store_result(%s) returned NULL, error %d (%s)", sql, mysql_errno(fMysql), mysql_error(fMysql));
      return DB_FILE_ERROR;
   }
 
   fNumFields = mysql_num_fields(fResult);
 
   //printf("num fields %d\n", fNumFields);
 
   return DB_SUCCESS;
}
 
int Mysql::Step()
{
   if (/* DISABLES CODE */ (0) && fDebug)
      printf("Mysql::Step()\n");
 
   assert(fMysql);
   assert(fResult);
 
   fRow = mysql_fetch_row(fResult);
 
   if (fRow)
      return DB_SUCCESS;
 
   if (mysql_errno(fMysql) == 0)
      return DB_NO_MORE_SUBKEYS;
 
   cm_msg(MERROR, "Mysql::Step", "mysql_fetch_row() error %d (%s)", mysql_errno(fMysql), mysql_error(fMysql));
 
   return DB_FILE_ERROR;
}
 
const char* Mysql::GetText(int column)
{
   assert(fMysql);
   assert(fResult);
   assert(fRow);
   assert(fNumFields > 0);
   assert(column >= 0);
   assert(column < fNumFields);
   if (fRow[column] == NULL)
      return "";
   return fRow[column];
}
 
double Mysql::GetDouble(int column)
{
   return atof(GetText(column));
}
 
time_t Mysql::GetTime(int column)
{
   return strtoul(GetText(column), NULL, 0);
}
 
int Mysql::Finalize()
{
   assert(fMysql);
   assert(fResult);
 
   mysql_free_result(fResult);
   fResult = NULL;
   fRow = NULL;
   fNumFields = 0;
 
   return DB_SUCCESS;
}
 
const char* Mysql::ColumnType(int midas_tid)
{
   assert(midas_tid>=0);
   assert(midas_tid<TID_LAST);
   return sql_type_mysql[midas_tid];
}
 
bool Mysql::TypesCompatible(int midas_tid, const char* sql_type)
{
   if (/* DISABLES CODE */ (0))
      printf("compare types midas \'%s\'=\'%s\' and sql \'%s\'\n", rpc_tid_name(midas_tid), ColumnType(midas_tid), sql_type);
 
   //if (sql2midasType_mysql(sql_type) == midas_tid)
   //   return true;
 
   if (strcasecmp(ColumnType(midas_tid), sql_type) == 0)
      return true;
 
   // permit writing FLOAT into DOUBLE
   if (midas_tid==TID_FLOAT && strcmp(sql_type, "double")==0)
      return true;
 
   // T2K quirk!
   // permit writing BYTE into signed tinyint
   if (midas_tid==TID_BYTE && strcmp(sql_type, "tinyint")==0)
      return true;
 
   // T2K quirk!
   // permit writing WORD into signed tinyint
   if (midas_tid==TID_WORD && strcmp(sql_type, "tinyint")==0)
      return true;
 
   // mysql quirk!
   //if (midas_tid==TID_DWORD && strcmp(sql_type, "int(10) unsigned")==0)
   //   return true;
 
   if (/* DISABLES CODE */ (0))
      printf("type mismatch!\n");
 
   return false;
}
 
std::string Mysql::QuoteId(const char* s)
{
   std::string q;
   q += "`";
   q += s;
   q += "`";
   return q;
}
 
std::string Mysql::QuoteString(const char* s)
{
   std::string q;
   q += "\'";
   q += s;
#if 0
   while (int c = *s++) {
      if (c == '\'') {
         q += "\\'";
      } if (c == '"') {
         q += "\\\"";
      } else if (isprint(c)) {
         q += c;
      } else {
         char buf[256];
         sprintf(buf, "\\\\x%02x", c&0xFF);
         q += buf;
      }
   }
#endif
   q += "\'";
   return q;
}
 
#endif // HAVE_MYSQL
 
#ifdef HAVE_PGSQL
 
//     PostgreSQL    database access  //
 
//#warning !!!HAVE_PGSQL!!!
 
#include <libpq-fe.h>
 
class Pgsql: public SqlBase
{
public:
   std::string fConnectString;
   int fDownsample = 0;
   PGconn* fPgsql = NULL;
 
   // query results
   PGresult *fResult = NULL;
   int fNumFields = 0;
   int fRow = 0;
 
   // disconnected operation
   size_t fMaxDisconnected = 0;
   std::list<std::string> fDisconnectedBuffer;
   time_t fNextReconnect = 0;
   int fNextReconnectDelaySec = 0;
   int fDisconnectedLost = 0;
 
   Pgsql(); // ctor
   ~Pgsql(); // dtor
 
   int Connect(const char* path);
   int Disconnect();
   bool IsConnected();
 
   int ConnectTable(const char* table_name);
 
   int ListTables(std::vector<std::string> *plist);
   int ListColumns(const char* table_name, std::vector<std::string> *plist);
 
   int Exec(const char* table_name, const char* sql);
   int ExecDisconnected(const char* table_name, const char* sql);
 
   int Prepare(const char* table_name, const char* sql);
   std::string BuildDownsampleQuery(const time_t start_time, const time_t end_time, const int npoints, const char* table_name, const char* column_name);
   int Step();
   const char* GetText(int column);
   time_t      GetTime(int column);
   double      GetDouble(int column);
   int Finalize();
 
   int OpenTransaction(const char* table_name);
   int CommitTransaction(const char* table_name);
   int RollbackTransaction(const char* table_name);
 
   const char* ColumnType(int midas_tid);
   bool TypesCompatible(int midas_tid, const char* sql_type);
 
   std::string QuoteId(const char* s);
   std::string QuoteString(const char* s);
};
 
Pgsql::Pgsql() // ctor
{
   fPgsql = NULL;
   fDownsample = 0;
   fResult = NULL;
   fRow = -1;
   fNumFields = 0;
   fMaxDisconnected = 1000;
   fNextReconnect = 0;
   fNextReconnectDelaySec = 0;
   fDisconnectedLost = 0;
   fTransactionPerTable = false;
}
 
Pgsql::~Pgsql() // dtor
{
   Disconnect();
   if(fResult)
      PQclear(fResult);
   fRow = -1;
   fNumFields = 0;
   if (fDisconnectedBuffer.size() > 0) {
      cm_msg(MINFO, "Pgsql::~Pgsql", "Lost %zu history entries accumulated while disconnected from the database", fDisconnectedBuffer.size());
      cm_msg_flush_buffer();
   }
}
 
int Pgsql::Connect(const char* connect_string)
{
   if (fIsConnected)
      Disconnect();
 
   fConnectString = connect_string;
 
   if (fDebug) {
      cm_msg(MINFO, "Pgsql::Connect", "Connecting to PostgreSQL database specified by \'%s\'", connect_string);
      cm_msg_flush_buffer();
   }
 
   std::string host_name;
   std::string user_name;
   std::string user_password;
   std::string db_name;
   std::string tcp_port;
   std::string unix_socket;
   std::string buffer;
 
   FILE* fp = fopen(connect_string, "r");
   if (!fp) {
      cm_msg(MERROR, "Pgsql::Connect", "Cannot read PostgreSQL connection parameters from \'%s\', fopen() error %d (%s)", connect_string, errno, strerror(errno));
      return DB_FILE_ERROR;
   }
 
   while (1) {
      char buf[256];
      char* s = fgets(buf, sizeof(buf), fp);
      if (!s)
         break; // EOF
 
      char*ss;
      // kill trailing \n and \r
      ss = strchr(s, '\n');
      if (ss) *ss = 0;
      ss = strchr(s, '\r');
      if (ss) *ss = 0;
 
      //printf("line [%s]\n", s);
 
      if (strncasecmp(s, "server=", 7)==0)
         host_name = skip_spaces(s + 7);
      if (strncasecmp(s, "port=", 5)==0)
         tcp_port = skip_spaces(s + 5);
      if (strncasecmp(s, "database=", 9)==0)
         db_name = skip_spaces(s + 9);
      if (strncasecmp(s, "socket=", 7)==0)
         unix_socket = skip_spaces(s + 7);
      if (strncasecmp(s, "user=", 5)==0)
         user_name = skip_spaces(s + 5);
      if (strncasecmp(s, "password=", 9)==0)
         user_password = skip_spaces(s + 9);
      if (strncasecmp(s, "buffer=", 7)==0)
         buffer = skip_spaces(s + 7);
   }
 
   fclose(fp);
 
   int buffer_int = atoi(buffer.c_str());
 
   if (buffer_int > 0 && buffer_int < 1000000)
      fMaxDisconnected = buffer_int;
 
   if (fDebug)
      printf("Pgsql::Connect: connecting to server [%s] port %s, unix socket [%s], database [%s], user [%s], password [%s], buffer [%zu]\n", host_name.c_str(), tcp_port.c_str(), unix_socket.c_str(), db_name.c_str(), user_name.c_str(), user_password.c_str(), fMaxDisconnected);
 
   fPgsql = PQsetdbLogin(host_name.c_str(), tcp_port.c_str(), NULL, NULL, db_name.c_str(), user_name.c_str(), user_password.c_str());
   if (PQstatus(fPgsql) != CONNECTION_OK) {
      std::string msg(PQerrorMessage(fPgsql));
      msg.erase(std::remove(msg.begin(), msg.end(), '\n'), msg.end());
      cm_msg(MERROR, "Pgsql::Connect", "PQsetdbLogin() to host [%s], port %s, unix socket [%s], database [%s], user [%s], password [%s]: error (%s)", host_name.c_str(), tcp_port.c_str(), unix_socket.c_str(), db_name.c_str(), user_name.c_str(), "xxx", msg.c_str());
      Disconnect();
      return DB_FILE_ERROR;
   }
 
   int status;
 
   if (fDebug) {
      cm_msg(MINFO, "Pgsql::Connect", "Connected to a PostgreSQL database on host [%s], port %s, unix socket [%s], database [%s], user [%s], password [%s], buffer %zu", host_name.c_str(), tcp_port.c_str(), unix_socket.c_str(), db_name.c_str(), user_name.c_str(), "xxx", fMaxDisconnected);
      cm_msg_flush_buffer();
   }
 
   fIsConnected = true;
 
   int count = 0;
   while (fDisconnectedBuffer.size() > 0) {
      status = Exec("(flush)", fDisconnectedBuffer.front().c_str());
      if (status != DB_SUCCESS) {
         return status;
      }
      fDisconnectedBuffer.pop_front();
      count++;
   }
 
   if (count > 0) {
      cm_msg(MINFO, "Pgsql::Connect", "Saved %d, lost %d history events accumulated while disconnected from the database", count, fDisconnectedLost);
      cm_msg_flush_buffer();
   }
 
   assert(fDisconnectedBuffer.size() == 0);
   fDisconnectedLost = 0;
 
   if (fDownsample) {
      status = Prepare("pg_extensions", "select extname from pg_extension where extname = 'timescaledb';");
 
      if (status != DB_SUCCESS || PQntuples(fResult) == 0) {
         cm_msg(MERROR, "Pgsql::Connect", "TimescaleDB extension not installed");
         return DB_FILE_ERROR;
      }
      Finalize();
 
      status = Prepare("pg_extensions", "select extname from pg_extension where extname = 'timescaledb_toolkit';");
 
      if (status != DB_SUCCESS || PQntuples(fResult) == 0) {
         cm_msg(MERROR, "Pgsql::Connect", "TimescaleDB_toolkit extension not installed");
         return DB_FILE_ERROR;
      }
      Finalize();
 
      cm_msg(MINFO, "Pgsql::Connect", "TimescaleDB extensions found - downsampling enabled");
   }
 
   return DB_SUCCESS;
}
 
int Pgsql::Disconnect()
{
   if (fPgsql)
      PQfinish(fPgsql);
 
   fPgsql = NULL;
   fRow = -1;
 
   fIsConnected = false;
   return DB_SUCCESS;
}
 
bool Pgsql::IsConnected()
{
   return fIsConnected;
}
 
int Pgsql::OpenTransaction(const char* table_name)
{
   return Exec(table_name, "BEGIN TRANSACTION;");
}
 
int Pgsql::CommitTransaction(const char* table_name)
{
   return Exec(table_name, "COMMIT;");
}
 
int Pgsql::RollbackTransaction(const char* table_name)
{
   return Exec(table_name, "ROLLBACK;");
}
 
int Pgsql::ListTables(std::vector<std::string> *plist)
{
   if (!fIsConnected)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Pgsql::ListTables!\n");
 
   int status = Prepare("pg_tables", "select tablename from pg_tables where schemaname = 'public';");
 
   if (status != DB_SUCCESS) {
      cm_msg(MERROR, "Pgsql::ListTables", "error %s (%s)", PQresStatus(PQresultStatus(fResult)), PQresultErrorMessage(fResult));
      return DB_FILE_ERROR;
   }
 
   while (1) {
      if (Step() != DB_SUCCESS)
         break;
      std::string tn = GetText(0);
      plist->push_back(tn);
   };
 
   Finalize();
 
   return DB_SUCCESS;
}
 
int Pgsql::ListColumns(const char* table_name, std::vector<std::string> *plist)
{
   if (!fIsConnected)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Pgsql::ListColumns for table \'%s\'\n", table_name);
 
   std::string cmd;
   cmd += "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = ";
   cmd += QuoteString(table_name);
   cmd += ";";
 
   int status = Prepare(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return status;
 
   fNumFields = PQnfields(fResult);
 
   while (1) {
      if (Step() != DB_SUCCESS)
         break;
      std::string cn = GetText(0);
      std::string ct = GetText(1);
      plist->push_back(cn);
      plist->push_back(ct);
   };
 
   Finalize();
 
   return DB_SUCCESS;
}
 
int Pgsql::Exec(const char* table_name, const char* sql)
{
   if (fDebug)
      printf("Pgsql::Exec(%s, %s)\n", table_name, sql);
 
   if (!fPgsql)
      return DB_FILE_ERROR;
 
   assert(fPgsql);
   assert(fRow == -1);
 
   fResult = PQexec(fPgsql, sql);
   ExecStatusType err = PQresultStatus(fResult);
   if(err != PGRES_TUPLES_OK) {
      if(err == PGRES_FATAL_ERROR) {
         // handle fatal error
         if(strstr(PQresultErrorMessage(fResult), "already exists"))
            return DB_KEY_EXIST;
         else return DB_FILE_ERROR;
      }
 
      if(PQstatus(fPgsql) == CONNECTION_BAD) {
         Disconnect();
         return ExecDisconnected(table_name, sql);
      }
   }
 
   return DB_SUCCESS;
}
 
int Pgsql::ExecDisconnected(const char* table_name, const char* sql)
{
   if (fDebug)
      printf("Pgsql::ExecDisconnected(%s, %s)\n", table_name, sql);
 
   if (fDisconnectedBuffer.size() < fMaxDisconnected) {
      fDisconnectedBuffer.push_back(sql);
      if (fDisconnectedBuffer.size() >= fMaxDisconnected) {
         cm_msg(MERROR, "Pgsql::ExecDisconnected", "Error: Disconnected database buffer overflow, size %zu, subsequent events are lost", fDisconnectedBuffer.size());
      }
   } else {
      fDisconnectedLost++;
   }
 
   time_t now = time(NULL);
 
   if (fNextReconnect == 0 || now >= fNextReconnect) {
      int status = Connect(fConnectString.c_str());
      if (status == DB_SUCCESS) {
         fNextReconnect = 0;
         fNextReconnectDelaySec = 0;
      } else {
         if (fNextReconnectDelaySec == 0) {
            fNextReconnectDelaySec = 5;
         } else if (fNextReconnectDelaySec < 10*60) {
            fNextReconnectDelaySec *= 2;
         }
         if (fDebug) {
            cm_msg(MINFO, "Pgsql::ExecDisconnected", "Next reconnect attempt in %d sec, history events buffered %zu, lost %d", fNextReconnectDelaySec, fDisconnectedBuffer.size(), fDisconnectedLost);
            cm_msg_flush_buffer();
         }
         fNextReconnect = now + fNextReconnectDelaySec;
      }
   }
   
   return DB_SUCCESS;
}
 
int Pgsql::Prepare(const char* table_name, const char* sql)
{
   if (fDebug)
      printf("Pgsql::Prepare(%s, %s)\n", table_name, sql);
 
   if (!fPgsql)
      return DB_FILE_ERROR;
 
   assert(fPgsql);
   //assert(fResult==NULL);
   assert(fRow == -1);
 
   fResult = PQexec(fPgsql, sql);
   if (PQstatus(fPgsql) == CONNECTION_BAD) {
      // lost connection to server
      int status = Connect(fConnectString.c_str());
      if (status == DB_SUCCESS) {
         // Retry after reconnecting
         fResult = PQexec(fPgsql, sql);
      } else {
         cm_msg(MERROR, "Pgsql::Prepare", "PQexec(%s) PostgreSQL server has gone away, and couldn't reconnect - %d", sql, status);
         return DB_FILE_ERROR;
      }
      if (status) {
         cm_msg(MERROR, "Pgsql::Prepare", "PQexec(%s) error %s", sql, PQresStatus(PQresultStatus(fResult)));
         return DB_FILE_ERROR;
      }
      cm_msg(MINFO, "Pgsql::Prepare", "Reconnected to PostgreSQL after long inactivity.");
   }
 
   fNumFields = PQnfields(fResult);
 
   return DB_SUCCESS;
}
 
std::string Pgsql::BuildDownsampleQuery(const time_t start_time, const time_t end_time, const int npoints, 
                                          const char* table_name, const char* column_name)
{
   std::string cmd;
   cmd += "SELECT extract(epoch from time::TIMESTAMPTZ) as _i_time, value ";
 
   cmd += " FROM unnest(( SELECT lttb";
   cmd += "(_t_time, ";
   cmd += column_name;
   cmd += ", ";
   cmd += std::to_string(npoints);
   cmd += ") ";
   cmd += "FROM ";
   cmd += QuoteId(table_name);
   cmd += " WHERE _t_time BETWEEN ";
   cmd += "to_timestamp(";
   cmd += TimeToString(start_time);
   cmd += ") AND to_timestamp(";
   cmd += TimeToString(end_time);
   cmd += ") )) ORDER BY time;";
 
   return cmd;
}
 
int Pgsql::Step()
{
   assert(fPgsql);
   assert(fResult);
 
   fRow++;
 
   if (fRow == PQntuples(fResult))
      return DB_NO_MORE_SUBKEYS;
 
   return DB_SUCCESS;
}
 
const char* Pgsql::GetText(int column)
{
   assert(fPgsql);
   assert(fResult);
   assert(fNumFields > 0);
   assert(column >= 0);
   assert(column < fNumFields);
 
   return PQgetvalue(fResult, fRow, column);
}
 
double Pgsql::GetDouble(int column)
{
   return atof(GetText(column));
}
 
time_t Pgsql::GetTime(int column)
{
   return strtoul(GetText(column), NULL, 0);
}
 
int Pgsql::Finalize()
{
   assert(fPgsql);
   assert(fResult);
 
   fRow = -1;
   fNumFields = 0;
 
   return DB_SUCCESS;
}
 
const char* Pgsql::ColumnType(int midas_tid)
{
   assert(midas_tid>=0);
   assert(midas_tid<TID_LAST);
   return sql_type_pgsql[midas_tid];
}
 
bool Pgsql::TypesCompatible(int midas_tid, const char* sql_type)
{
   if (/* DISABLES CODE */ (0))
      printf("compare types midas \'%s\'=\'%s\' and sql \'%s\'\n", rpc_tid_name(midas_tid), ColumnType(midas_tid), sql_type);
 
   //if (sql2midasType_mysql(sql_type) == midas_tid)
   //   return true;
 
   if (strcasecmp(ColumnType(midas_tid), sql_type) == 0)
      return true;
 
   // permit writing FLOAT into DOUBLE
   if (midas_tid==TID_FLOAT && strcmp(sql_type, "double precision")==0)
      return true;
 
   // T2K quirk!
   // permit writing BYTE into signed tinyint
   if (midas_tid==TID_BYTE && strcmp(sql_type, "integer")==0)
      return true;
 
   // T2K quirk!
   // permit writing WORD into signed tinyint
   if (midas_tid==TID_WORD && strcmp(sql_type, "integer")==0)
      return true;
 
   if (/* DISABLES CODE */ (0))
      printf("type mismatch!\n");
 
   return false;
}
 
std::string Pgsql::QuoteId(const char* s)
{
   std::string q;
   q += '"';
   q += s;
   q += '"';
   return q;
}
 
std::string Pgsql::QuoteString(const char* s)
{
   std::string q;
   q += '\'';
   q += s;
   q += '\'';
   return q;
}
 
#endif // HAVE_PGSQL
 
#ifdef HAVE_SQLITE
 
//        SQLITE database access      //
 
#include <sqlite3.h>
 
typedef std::map<std::string, sqlite3*> DbMap;
 
class Sqlite: public SqlBase
{
public:
   std::string fPath;
 
   DbMap fMap;
 
   // temporary storage of query data
   sqlite3* fTempDB;
   sqlite3_stmt* fTempStmt;
 
   Sqlite(); // ctor
   ~Sqlite(); // dtor
 
   int Connect(const char* path);
   int Disconnect();
   bool IsConnected();
 
   int ConnectTable(const char* table_name);
   sqlite3* GetTable(const char* table_name);
 
   int ListTables(std::vector<std::string> *plist);
   int ListColumns(const char* table_name, std::vector<std::string> *plist);
 
   int Exec(const char* table_name, const char* sql);
   int ExecDisconnected(const char* table_name, const char* sql);
 
   int Prepare(const char* table_name, const char* sql);
   int Step();
   const char* GetText(int column);
   time_t      GetTime(int column);
   double      GetDouble(int column);
   int Finalize();
 
   int OpenTransaction(const char* table_name);
   int CommitTransaction(const char* table_name);
   int RollbackTransaction(const char* table_name);
 
   const char* ColumnType(int midas_tid);
   bool TypesCompatible(int midas_tid, const char* sql_type);
 
   std::string QuoteId(const char* s);
   std::string QuoteString(const char* s);
};
 
std::string Sqlite::QuoteId(const char* s)
{
   std::string q;
   q += "\"";
   q += s;
   q += "\"";
   return q;
}
 
std::string Sqlite::QuoteString(const char* s)
{
   std::string q;
   q += "\'";
   q += s;
   q += "\'";
   return q;
}
 
const char* Sqlite::ColumnType(int midas_tid)
{
   assert(midas_tid>=0);
   assert(midas_tid<TID_LAST);
   return sql_type_sqlite[midas_tid];
}
 
bool Sqlite::TypesCompatible(int midas_tid, const char* sql_type)
{
   if (0)
      printf("compare types midas \'%s\'=\'%s\' and sql \'%s\'\n", rpc_tid_name(midas_tid), ColumnType(midas_tid), sql_type);
 
   //if (sql2midasType_sqlite(sql_type) == midas_tid)
   //   return true;
 
   if (strcasecmp(ColumnType(midas_tid), sql_type) == 0)
      return true;
 
   // permit writing FLOAT into DOUBLE
   if (midas_tid==TID_FLOAT && strcasecmp(sql_type, "double")==0)
      return true;
 
   return false;
}
 
const char* Sqlite::GetText(int column)
{
   return (const char*)sqlite3_column_text(fTempStmt, column);
}
 
time_t Sqlite::GetTime(int column)
{
   return sqlite3_column_int64(fTempStmt, column);
}
 
double Sqlite::GetDouble(int column)
{
   return sqlite3_column_double(fTempStmt, column);
}
 
Sqlite::Sqlite() // ctor
{
   fIsConnected = false;
   fTempDB = NULL;
   fTempStmt = NULL;
   fDebug = 0;
}
 
Sqlite::~Sqlite() // dtor
{
   Disconnect();
}
 
const char* xsqlite3_errstr(sqlite3* db, int errcode)
{
   //return sqlite3_errstr(errcode);
   return sqlite3_errmsg(db);
}
 
int Sqlite::ConnectTable(const char* table_name)
{
   std::string fname = fPath + "mh_" + table_name + ".sqlite3";
 
   sqlite3* db = NULL;
 
   int status = sqlite3_open(fname.c_str(), &db);
 
   if (status != SQLITE_OK) {
      cm_msg(MERROR, "Sqlite::Connect", "Table %s: sqlite3_open(%s) error %d (%s)", table_name, fname.c_str(), status, xsqlite3_errstr(db, status));
      sqlite3_close(db);
      db = NULL;
      return DB_FILE_ERROR;
   }
 
#if SQLITE_VERSION_NUMBER >= 3006020
   status = sqlite3_extended_result_codes(db, 1);
   if (status != SQLITE_OK) {
      cm_msg(MERROR, "Sqlite::Connect", "Table %s: sqlite3_extended_result_codes(1) error %d (%s)", table_name, status, xsqlite3_errstr(db, status));
   }
#else
#warning Missing sqlite3_extended_result_codes()!
#endif
 
   fMap[table_name] = db;
 
   Exec(table_name, "PRAGMA journal_mode=persist;");
   Exec(table_name, "PRAGMA synchronous=normal;");
   //Exec(table_name, "PRAGMA synchronous=off;");
   Exec(table_name, "PRAGMA journal_size_limit=-1;");
 
   if (0) {
      Exec(table_name, "PRAGMA legacy_file_format;");
      Exec(table_name, "PRAGMA synchronous;");
      Exec(table_name, "PRAGMA journal_mode;");
      Exec(table_name, "PRAGMA journal_size_limit;");
   }
 
#ifdef SQLITE_LIMIT_COLUMN
   if (0) {
      int max_columns = sqlite3_limit(db, SQLITE_LIMIT_COLUMN, -1);
      printf("Sqlite::Connect: SQLITE_LIMIT_COLUMN=%d\n", max_columns);
   }
#endif
 
   if (fDebug)
      cm_msg(MINFO, "Sqlite::Connect", "Table %s: connected to Sqlite file \'%s\'", table_name, fname.c_str());
 
   return DB_SUCCESS;
}
 
sqlite3* Sqlite::GetTable(const char* table_name)
{
   sqlite3* db = fMap[table_name];
 
   if (db)
      return db;
 
   int status = ConnectTable(table_name);
   if (status != DB_SUCCESS)
      return NULL;
 
   return fMap[table_name];
}
 
int Sqlite::Connect(const char* path)
{
   if (fIsConnected)
      Disconnect();
 
   fPath = path;
 
   // add trailing '/'
   if (fPath.length() > 0) {
      if (fPath[fPath.length()-1] != DIR_SEPARATOR)
         fPath += DIR_SEPARATOR_STR;
   }
 
   if (fDebug)
      cm_msg(MINFO, "Sqlite::Connect", "Connected to Sqlite database in \'%s\'", fPath.c_str());
 
   fIsConnected = true;
 
   return DB_SUCCESS;
}
 
int Sqlite::Disconnect()
{
   if (!fIsConnected)
      return DB_SUCCESS;
 
   for (DbMap::iterator iter = fMap.begin(); iter != fMap.end(); ++iter) {
      const char* table_name = iter->first.c_str();
      sqlite3* db = iter->second;
      int status = sqlite3_close(db);
      if (status != SQLITE_OK) {
         cm_msg(MERROR, "Sqlite::Disconnect", "sqlite3_close(%s) error %d (%s)", table_name, status, xsqlite3_errstr(db, status));
      }
   }
 
   fMap.clear();
 
   fIsConnected = false;
 
   return DB_SUCCESS;
}
 
bool Sqlite::IsConnected()
{
   return fIsConnected;
}
 
int Sqlite::OpenTransaction(const char* table_name)
{
   int status = Exec(table_name, "BEGIN TRANSACTION");
   return status;
}
 
int Sqlite::CommitTransaction(const char* table_name)
{
   int status = Exec(table_name, "COMMIT TRANSACTION");
   return status;
}
 
int Sqlite::RollbackTransaction(const char* table_name)
{
   int status = Exec(table_name, "ROLLBACK TRANSACTION");
   return status;
}
 
int Sqlite::Prepare(const char* table_name, const char* sql)
{
   sqlite3* db = GetTable(table_name);
   if (!db)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Sqlite::Prepare(%s, %s)\n", table_name, sql);
 
   assert(fTempDB==NULL);
   fTempDB = db;
 
#if SQLITE_VERSION_NUMBER >= 3006020
   int status = sqlite3_prepare_v2(db, sql, strlen(sql), &fTempStmt, NULL);
#else
#warning Missing sqlite3_prepare_v2()!
   int status = sqlite3_prepare(db, sql, strlen(sql), &fTempStmt, NULL);
#endif
 
   if (status == SQLITE_OK)
      return DB_SUCCESS;
 
   std::string sqlstring = sql;
   cm_msg(MERROR, "Sqlite::Prepare", "Table %s: sqlite3_prepare_v2(%s...) error %d (%s)", table_name, sqlstring.substr(0,60).c_str(), status, xsqlite3_errstr(db, status));
 
   fTempDB = NULL;
 
   return DB_FILE_ERROR;
}
 
int Sqlite::Step()
{
   if (0 && fDebug)
      printf("Sqlite::Step()\n");
 
   assert(fTempDB);
   assert(fTempStmt);
 
   int status = sqlite3_step(fTempStmt);
 
   if (status == SQLITE_DONE)
      return DB_NO_MORE_SUBKEYS;
 
   if (status == SQLITE_ROW)
      return DB_SUCCESS;
 
   cm_msg(MERROR, "Sqlite::Step", "sqlite3_step() error %d (%s)", status, xsqlite3_errstr(fTempDB, status));
 
   return DB_FILE_ERROR;
}
 
int Sqlite::Finalize()
{
   if (0 && fDebug)
      printf("Sqlite::Finalize()\n");
 
   assert(fTempDB);
   assert(fTempStmt);
 
   int status = sqlite3_finalize(fTempStmt);
 
   if (status != SQLITE_OK) {
      cm_msg(MERROR, "Sqlite::Finalize", "sqlite3_finalize() error %d (%s)", status, xsqlite3_errstr(fTempDB, status));
 
      fTempDB = NULL;
      fTempStmt = NULL; // FIXME: maybe a memory leak?
      return DB_FILE_ERROR;
   }
 
   fTempDB = NULL;
   fTempStmt = NULL;
 
   return DB_SUCCESS;
}
 
int Sqlite::ListTables(std::vector<std::string> *plist)
{
   if (!fIsConnected)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Sqlite::ListTables at path [%s]\n", fPath.c_str());
 
   int status;
 
   const char* cmd = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name;";
 
   DIR *dir = opendir(fPath.c_str());
   if (!dir) {
      cm_msg(MERROR, "Sqlite::ListTables", "Cannot opendir(%s), errno %d (%s)", fPath.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   while (1) {
      const struct dirent* de = readdir(dir);
      if (!de)
         break;
 
      const char* dn = de->d_name;
 
      //if (dn[0]!='m' || dn[1]!='h')
      //continue;
 
      const char* s;
 
      s = strstr(dn, "mh_");
      if (!s || s!=dn)
         continue;
 
      s = strstr(dn, ".sqlite3");
      if (!s || s[8]!=0)
         continue;
 
      char table_name[256];
      mstrlcpy(table_name, dn+3, sizeof(table_name));
      // FIXME: skip names like "xxx.sqlite3~" and "xxx.sqlite3-deleted"
      char* ss = strstr(table_name, ".sqlite3");
      if (!ss)
         continue;
      *ss = 0;
 
      //printf("dn [%s] tn [%s]\n", dn, table_name);
 
      status = Prepare(table_name, cmd);
      if (status != DB_SUCCESS)
         continue;
 
      while (1) {
         status = Step();
         if (status != DB_SUCCESS)
            break;
 
         const char* tn = GetText(0);
         //printf("table [%s]\n", tn);
         plist->push_back(tn);
      }
 
      status = Finalize();
   }
 
   closedir(dir);
   dir = NULL;
 
   return DB_SUCCESS;
}
 
int Sqlite::ListColumns(const char* table, std::vector<std::string> *plist)
{
   if (!fIsConnected)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Sqlite::ListColumns for table \'%s\'\n", table);
 
   std::string cmd;
   cmd = "PRAGMA table_info(";
   cmd += table;
   cmd += ");";
 
   int status;
 
   status = Prepare(table, cmd.c_str());
   if (status != DB_SUCCESS)
      return status;
 
   while (1) {
      status = Step();
      if (status != DB_SUCCESS)
         break;
 
      const char* colname = GetText(1);
      const char* coltype = GetText(2);
      //printf("column [%s] [%s]\n", colname, coltype);
      plist->push_back(colname); // column name
      plist->push_back(coltype); // column type
   }
 
   status = Finalize();
 
   return DB_SUCCESS;
}
 
static int callback_debug = 0;
 
static int callback(void *NotUsed, int argc, char **argv, char **azColName){
   if (callback_debug) {
      printf("history_sqlite::callback---->\n");
      for (int i=0; i<argc; i++){
         printf("history_sqlite::callback[%d] %s = %s\n", i, azColName[i], argv[i] ? argv[i] : "NULL");
      }
   }
   return 0;
}
 
int Sqlite::Exec(const char* table_name, const char* sql)
{
   // return values:
   // DB_SUCCESS
   // DB_FILE_ERROR: not connected
   // DB_KEY_EXIST: "table already exists"
 
   if (!fIsConnected)
      return DB_FILE_ERROR;
 
   sqlite3* db = GetTable(table_name);
   if (!db)
      return DB_FILE_ERROR;
 
   if (fDebug)
      printf("Sqlite::Exec(%s, %s)\n", table_name, sql);
 
   int status;
 
   callback_debug = fDebug;
   char* errmsg = NULL;
 
   status = sqlite3_exec(db, sql, callback, 0, &errmsg);
   if (status != SQLITE_OK) {
      if (status == SQLITE_ERROR && strstr(errmsg, "duplicate column name"))
         return DB_KEY_EXIST;
      if (status == SQLITE_ERROR && strstr(errmsg, "already exists"))
         return DB_KEY_EXIST;
      std::string sqlstring = sql;
      cm_msg(MERROR, "Sqlite::Exec", "Table %s: sqlite3_exec(%s...) error %d (%s)", table_name, sqlstring.substr(0,60).c_str(), status, errmsg);
      sqlite3_free(errmsg);
      return DB_FILE_ERROR;
   }
 
   return DB_SUCCESS;
}
 
int Sqlite::ExecDisconnected(const char* table_name, const char* sql)
{
   cm_msg(MERROR, "Sqlite::Exec", "sqlite driver does not support disconnected operations");
   return DB_FILE_ERROR;
}
 
#endif // HAVE_SQLITE
 
//    Methods of HsFileSchema     //
 
int HsFileSchema::write_event(const time_t t, const char* data, const size_t data_size)
{
   HsFileSchema* s = this;
 
   assert(s->fVariables.size() == s->fOffsets.size());
 
   if (s->fWriterFd < 0) {
      s->fWriterFd = open(s->fFileName.c_str(), O_RDWR);
      if (s->fWriterFd < 0) {
         cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', open() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
         return HS_FILE_ERROR;
      }
 
      //static_assert((sizeof(int)==3),"here!");
      //static_assert((sizeof(off64_t)==7),"here!");
      //static_assert((sizeof(size_t)==3),"here!");
      //static_assert((sizeof(intmax_t)==3),"here!");
 
      //off64_t zzz = 1;
      //off64_t xxx = zzz<<60;
      //int yyy = xxx;
      //printf("%d", yyy);
 
      off64_t file_size = ::lseek64(s->fWriterFd, 0, SEEK_END);
 
      if (file_size < 0) {
         cm_msg(MERROR, "FileHistory::write_event", "Cannot read file size of \'%s\', lseek64(SEEK_END) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
         return HS_FILE_ERROR;
      }
 
      off64_t nrec = 0;
 
      if (file_size >= s->fDataOffset) {
         nrec = (file_size - s->fDataOffset)/s->fRecordSize;
      }
 
      if (nrec < 0)
         nrec = 0;
 
      off64_t data_end = s->fDataOffset + nrec*s->fRecordSize;
      
      //printf("write_event: file_size %jd, nrec %jd, data_end %jd\n", (intmax_t)file_size, (intmax_t)nrec, (intmax_t)data_end);
 
      if (data_end != file_size) {
         if (nrec > 0)
            cm_msg(MERROR, "FileHistory::write_event", "File \'%s\' may be truncated, data offset %jd, record size %zu, file size: %jd, should be %jd, making it so", s->fFileName.c_str(), (intmax_t)s->fDataOffset, s->fRecordSize, (intmax_t)file_size, (intmax_t)data_end);
 
         off64_t status64 = ::lseek64(s->fWriterFd, data_end, SEEK_SET);
 
         if (status64 < 0) {
            cm_msg(MERROR, "FileHistory::write_event", "Cannot seek \'%s\' to offset %jd, lseek64() errno %d (%s)", s->fFileName.c_str(), (intmax_t)data_end, errno, strerror(errno));
            return HS_FILE_ERROR;
         }
 
         int status = ::ftruncate64(s->fWriterFd, data_end);
 
         if (status < 0) {
            cm_msg(MERROR, "FileHistory::write_event", "Cannot truncate \'%s\' to size %jd, ftruncate64() errno %d (%s)", s->fFileName.c_str(), (intmax_t)data_end, errno, strerror(errno));
            return HS_FILE_ERROR;
         }
      }
 
      s->fFileSizeInitial = data_end;
      s->fFileSize = data_end;
   }
 
   size_t expected_size = s->fRecordSize - 4;
 
   // sanity check: record_size and n_bytes are computed from the byte counts in the file header
   assert(expected_size == s->fNumBytes);
 
   if (s->fLastSize == 0)
      s->fLastSize = expected_size;
 
   if (data_size != s->fLastSize) {
      cm_msg(MERROR, "FileHistory::write_event", "Event \'%s\' data size mismatch, expected %zu bytes, got %zu bytes, previously %zu bytes", s->fEventName.c_str(), expected_size, data_size, s->fLastSize);
      //printf("schema:\n");
      //s->print();
 
      if (data_size < expected_size)
         return HS_FILE_ERROR;
 
      // truncate for now
      // data_size = expected_size;
      s->fLastSize = data_size;
   }
 
   size_t size = 4 + expected_size;
 
   if (size != s->fRecordBufferSize) {
      s->fRecordBuffer = (char*)realloc(s->fRecordBuffer, size);
      assert(s->fRecordBuffer != NULL);
      s->fRecordBufferSize = size;
   }
 
   memcpy(s->fRecordBuffer, &t, 4);
   memcpy(s->fRecordBuffer+4, data, expected_size);
 
   ssize_t wr = write(s->fWriterFd, s->fRecordBuffer, size);
   if ((size_t)wr != size) {
      cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', write(%zu) returned %zd, errno %d (%s)", s->fFileName.c_str(), size, wr, errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   s->fFileSize += size;
 
#if 0
   status = write(s->fWriterFd, &t, 4);
   if (status != 4) {
      cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', write(timestamp) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   status = write(s->fWriterFd, data, expected_size);
   if (status != expected_size) {
      cm_msg(MERROR, "FileHistory::write_event", "Cannot write to \'%s\', write(%d) errno %d (%s)", s->fFileName.c_str(), data_size, errno, strerror(errno));
      return HS_FILE_ERROR;
   }
#endif
 
   return HS_SUCCESS;
}
 
int HsFileSchema::close()
{
   if (fWriterFd >= 0) {
      ::close(fWriterFd);
      fWriterFd = -1;
   }
   return HS_SUCCESS;
}
 
static int ReadRecord(const char* file_name, int fd, off64_t offset, size_t recsize, off64_t irec, char* rec)
{
   off64_t fpos = offset + irec*recsize;
 
   off64_t status64 = ::lseek64(fd, fpos, SEEK_SET);
 
   if (status64 < 0) {
      cm_msg(MERROR, "FileHistory::ReadRecord", "Cannot read \'%s\', lseek64(%jd) errno %d (%s)", file_name, (intmax_t)fpos, errno, strerror(errno));
      return -1;
   }
 
   ssize_t rd = ::read(fd, rec, recsize);
 
   if (rd < 0) {
      cm_msg(MERROR, "FileHistory::ReadRecord", "Cannot read \'%s\', read() errno %d (%s)", file_name, errno, strerror(errno));
      return -1;
   }
 
   if (rd == 0) {
      cm_msg(MERROR, "FileHistory::ReadRecord", "Cannot read \'%s\', unexpected end of file on read()", file_name);
      return -1;
   }
 
   if ((size_t)rd != recsize) {
      cm_msg(MERROR, "FileHistory::ReadRecord", "Cannot read \'%s\', short read() returned %zd instead of %zu bytes", file_name, rd, recsize);
      return -1;
   }
 
   return HS_SUCCESS;
}
 
static int FindTime(const char* file_name, int fd, off64_t offset, size_t recsize, off64_t nrec, time_t timestamp, off64_t* i1p, time_t* t1p, off64_t* i2p, time_t* t2p, time_t* tstart, time_t* tend, int debug)
{
   //
   // purpose: find location time timestamp inside given file.
   // uses binary search
   // returns:
   // tstart, tend - time of first and last data in a file
   // i1p,t1p - data just before timestamp, used as "last_written"
   // i2p,t2p - data at timestamp or after timestamp, used as starting point to read data from file
   // assertions:
   // tstart <= t1p < t2p <= tend
   // i1p+1==i2p
   // t1p < timestamp <= t2p
   //
   // special cases:
   // 1) timestamp <= tstart - all data is in the future, return i1p==-1, t1p==-1, i2p==0, t2p==tstart
   // 2) tend < timestamp - all the data is in the past, return i1p = nrec-1, t1p = tend, i2p = nrec, t2p = 0;
   // 3) nrec == 1 only one record in this file and it is older than the timestamp (tstart == tend < timestamp)
   //
 
   int status;
   char* buf = new char[recsize];
 
   assert(nrec > 0);
 
   off64_t rec1 = 0;
   off64_t rec2 = nrec-1;
 
   status = ReadRecord(file_name, fd, offset, recsize, rec1, buf);
   if (status != HS_SUCCESS) {
      delete[] buf;
      return HS_FILE_ERROR;
   }
 
   time_t t1 = *(DWORD*)buf;
 
   *tstart = t1;
 
   // timestamp is older than any data in this file
   if (timestamp <= t1) {
      *i1p    = -1;
      *t1p    =  0;
      *i2p    = 0;
      *t2p    = t1;
      *tend   = 0;
      delete[] buf;
      return HS_SUCCESS;
   }
 
   assert(t1 < timestamp);
 
   if (nrec == 1) {
      *i1p    = 0;
      *t1p    = t1;
      *i2p    = nrec; // == 1
      *t2p    = 0;
      *tend   = t1;
      delete[] buf;
      return HS_SUCCESS;
   }
 
   status = ReadRecord(file_name, fd, offset, recsize, rec2, buf);
   if (status != HS_SUCCESS) {
      delete[] buf;
      return HS_FILE_ERROR;
   }
 
   time_t t2 = *(DWORD*)buf;
 
   *tend = t2;
 
   // all the data is in the past
   if (t2 < timestamp) {
      *i1p = rec2;
      *t1p = t2;
      *i2p = nrec;
      *t2p = 0;
      delete[] buf;
      return HS_SUCCESS;
   }
 
   assert(t1 < timestamp);
   assert(timestamp <= t2);
 
   if (debug)
      printf("FindTime: rec %jd..(x)..%jd, time %s..(%s)..%s\n", (intmax_t)rec1, (intmax_t)rec2, TimeToString(t1).c_str(), TimeToString(timestamp).c_str(), TimeToString(t2).c_str());
 
   // implement binary search
 
   do {
      off64_t rec = (rec1+rec2)/2;
 
      assert(rec >= 0);
      assert(rec < nrec);
 
      status = ReadRecord(file_name, fd, offset, recsize, rec, buf);
      if (status != HS_SUCCESS) {
         delete[] buf;
         return HS_FILE_ERROR;
      }
 
      time_t t = *(DWORD*)buf;
 
      if (timestamp <= t) {
         if (debug)
            printf("FindTime: rec %jd..(x)..%jd..%jd, time %s..(%s)..%s..%s\n", (intmax_t)rec1, (intmax_t)rec, (intmax_t)rec2, TimeToString(t1).c_str(), TimeToString(timestamp).c_str(), TimeToString(t).c_str(), TimeToString(t2).c_str());
 
         rec2 = rec;
         t2 = t;
      } else {
         if (debug)
            printf("FindTime: rec %jd..%jd..(x)..%jd, time %s..%s..(%s)..%s\n", (intmax_t)rec1, (intmax_t)rec, (intmax_t)rec2, TimeToString(t1).c_str(), TimeToString(t).c_str(), TimeToString(timestamp).c_str(), TimeToString(t2).c_str());
 
         rec1 = rec;
         t1 = t;
      }
   } while (rec2 - rec1 > 1);
 
   assert(rec1+1 == rec2);
   assert(t1 < timestamp);
   assert(timestamp <= t2);
 
   if (debug)
      printf("FindTime: rec %jd..(x)..%jd, time %s..(%s)..%s, this is the result.\n", (intmax_t)rec1, (intmax_t)rec2, TimeToString(t1).c_str(), TimeToString(timestamp).c_str(), TimeToString(t2).c_str());
 
   *i1p = rec1;
   *t1p = t1;
 
   *i2p = rec2;
   *t2p = t2;
 
   delete[] buf;
   return HS_SUCCESS;
}
 
int HsFileSchema::read_last_written(const time_t timestamp,
                                    const int debug,
                                    time_t* last_written)
{
   int status;
   HsFileSchema* s = this;
 
   if (debug)
      printf("FileHistory::read_last_written: file %s, schema time %s..%s, timestamp %s\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(timestamp).c_str());
 
   int fd = open(s->fFileName.c_str(), O_RDONLY);
   if (fd < 0) {
      cm_msg(MERROR, "FileHistory::read_last_written", "Cannot read \'%s\', open() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   off64_t file_size = ::lseek64(fd, 0, SEEK_END);
 
   if (file_size < 0) {
      cm_msg(MERROR, "FileHistory::read_last_written", "Cannot read file size of \'%s\', lseek64(SEEK_END) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   if (file_size < s->fDataOffset) {
      // empty file
      ::close(fd);
      if (last_written)
         *last_written = 0;
      return HS_SUCCESS;
   }
 
   off64_t nrec = (file_size - s->fDataOffset)/s->fRecordSize;
 
   //printf("read_last_written: nrec %jd, file_size %jd, offset %jd, record size %zu\n", (intmax_t)nrec, (intmax_t)file_size, s->fDataOffset, s->fRecordSize);
 
   if (nrec < 0)
      nrec = 0;
 
   if (nrec < 1) {
      ::close(fd);
      if (last_written)
         *last_written = 0;
      return HS_SUCCESS;
   }
 
   time_t lw = 0;
 
   // read last record to check if desired time is inside or outside of the file
 
   if (1) {
      char* buf = new char[s->fRecordSize];
 
      status = ReadRecord(s->fFileName.c_str(), fd, s->fDataOffset, s->fRecordSize, nrec - 1, buf);
      if (status != HS_SUCCESS) {
         delete[] buf;
         ::close(fd);
         return HS_FILE_ERROR;
      }
 
      lw = *(DWORD*)buf;
 
      delete[] buf;
   }
 
   if (lw >= timestamp) {
      off64_t irec = 0;
      time_t trec = 0;
      off64_t iunused = 0;
      time_t tunused = 0;
      time_t tstart = 0; // not used
      time_t tend   = 0; // not used
 
      status = FindTime(s->fFileName.c_str(), fd, s->fDataOffset, s->fRecordSize, nrec, timestamp, &irec, &trec, &iunused, &tunused, &tstart, &tend, 0*debug);
      if (status != HS_SUCCESS) {
         ::close(fd);
         return HS_FILE_ERROR;
      }
 
      assert(trec < timestamp);
 
      lw = trec;
   }
 
   if (last_written)
      *last_written = lw;
 
   if (debug)
      printf("FileHistory::read_last_written: file %s, schema time %s..%s, timestamp %s, last_written %s\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(timestamp).c_str(), TimeToString(lw).c_str());
 
   assert(lw < timestamp);
 
   ::close(fd);
 
   return HS_SUCCESS;
}
 
int HsFileSchema::read_data(const time_t start_time,
                            const time_t end_time,
                            const int num_var, const std::vector<int>& var_schema_index, const int var_index[],
                            const int debug,
                            std::vector<time_t>& last_time,
                            MidasHistoryBufferInterface* buffer[])
{
   HsFileSchema* s = this;
 
   if (debug)
      printf("FileHistory::read_data: file %s, schema time %s..%s, read time %s..%s, %d vars\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), num_var);
 
   //if (1) {
   //   printf("Last time: ");
   //   for (int i=0; i<num_var; i++) {
   //      printf(" %s", TimeToString(last_time[i]).c_str());
   //   }
   //   printf("\n");
   //}
 
   if (debug) {
      printf("FileHistory::read_data: file %s map", s->fFileName.c_str());
      for (size_t i=0; i<var_schema_index.size(); i++) {
         printf(" %2d", var_schema_index[i]);
      }
      printf("\n");
   }
 
   int fd = ::open(s->fFileName.c_str(), O_RDONLY);
   if (fd < 0) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', open() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   off64_t file_size = ::lseek64(fd, 0, SEEK_END);
 
   if (file_size < 0) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read file size of \'%s\', lseek64(SEEK_END) errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   if (file_size < s->fDataOffset) {
      // empty file
      ::close(fd);
      return HS_SUCCESS;
   }
 
   off64_t nrec = (file_size - s->fDataOffset)/s->fRecordSize;
 
   //printf("read_data: nrec %jd, file_size %jd, offset %jd, record size %zu\n", (intmax_t)nrec, (intmax_t)file_size, s->fDataOffset, s->fRecordSize);
 
   if (nrec < 0)
      nrec = 0;
 
   if (nrec < 1) {
      ::close(fd);
      return HS_SUCCESS;
   }
 
   off64_t iunused = 0;
   time_t tunused = 0;
   off64_t irec = 0;
   time_t trec   = 0;
   time_t tstart = 0;
   time_t tend   = 0;
 
   int istatus = FindTime(s->fFileName.c_str(), fd, s->fDataOffset, s->fRecordSize, nrec, start_time, &iunused, &tunused, &irec, &trec, &tstart, &tend, 0*debug);
 
   if (istatus != HS_SUCCESS) {
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   if (debug) {
      printf("FindTime %d, nrec %jd, (%jd, %s) (%jd, %s), tstart %s, tend %s, want %s\n", istatus, (intmax_t)nrec, (intmax_t)iunused, TimeToString(tunused).c_str(), (intmax_t)irec, TimeToString(trec).c_str(), TimeToString(tstart).c_str(), TimeToString(tend).c_str(), TimeToString(start_time).c_str());
   }
 
   if (irec < 0 || irec >= nrec) {
      // all data in this file is older than start_time
 
      ::close(fd);
 
      if (debug)
         printf("FileHistory::read: file %s, schema time %s..%s, read time %s..%s, file time %s..%s, data in this file is too old\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), TimeToString(tstart).c_str(), TimeToString(tend).c_str());
 
      return HS_SUCCESS;
   }
 
   if (tstart < s->fTimeFrom) {
      // data starts before time declared in schema
 
      ::close(fd);
 
      cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': timestamp of first data %s is before schema start time %s", s->fFileName.c_str(), TimeToString(tstart).c_str(), TimeToString(s->fTimeFrom).c_str());
 
      return HS_FILE_ERROR;
   }
 
   if (tend && s->fTimeTo && tend > s->fTimeTo) {
      // data ends after time declared in schema (overlaps with next file)
 
      ::close(fd);
 
      cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': timestamp of last data %s is after schema end time %s", s->fFileName.c_str(), TimeToString(tend).c_str(), TimeToString(s->fTimeTo).c_str());
 
      return HS_FILE_ERROR;
   }
 
   for (int i=0; i<num_var; i++) {
      int si = var_schema_index[i];
      if (si < 0)
         continue;
         
      if (trec < last_time[i]) { // protect against duplicate and non-monotonous data
         ::close(fd);
 
         cm_msg(MERROR, "FileHistory::read_data", "Internal history error at file \'%s\': variable %d data timestamp %s is before last timestamp %s", s->fFileName.c_str(), i, TimeToString(trec).c_str(), TimeToString(last_time[i]).c_str());
         
         return HS_FILE_ERROR;
      }
   }
 
   int count = 0;
 
   off64_t fpos = s->fDataOffset + irec*s->fRecordSize;
   
   off64_t xpos = ::lseek64(fd, fpos, SEEK_SET);
 
   //printf("read_data: lseek64() returned %jd, irec %jd, offset %jd, record size %zu, fpos %jd\n", xpos, irec, s->fDataOffset, s->fRecordSize, fpos);
 
   if (xpos < 0) {
      cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', lseek64(%jd) errno %d (%s)", s->fFileName.c_str(), (intmax_t)fpos, errno, strerror(errno));
      ::close(fd);
      return HS_FILE_ERROR;
   }
 
   char* buf = new char[s->fRecordSize];
 
   off64_t prec = irec;
 
   while (1) {
      ssize_t rd = ::read(fd, buf, s->fRecordSize);
 
      if (rd < 0) {
         cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', read() errno %d (%s)", s->fFileName.c_str(), errno, strerror(errno));
         break;
      }
 
      if (rd == 0) {
         // EOF
         break;
      }
 
      if ((size_t)rd != s->fRecordSize) {
         cm_msg(MERROR, "FileHistory::read_data", "Cannot read \'%s\', short read() returned %zd instead of %zu bytes", s->fFileName.c_str(), rd, s->fRecordSize);
         break;
      }
 
      prec++;
 
      bool past_end_of_last_file = (s->fTimeTo == 0) && (prec > nrec);
 
      time_t t = *(DWORD*)buf;
      
      if (debug > 1)
         printf("FileHistory::read: file %s, schema time %s..%s, read time %s..%s, row time %s\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), TimeToString(t).c_str());
      
      if (t < trec) {
         delete[] buf;
         ::close(fd);
         cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': record %jd timestamp %s is before start time %s", s->fFileName.c_str(), (intmax_t)(irec + count), TimeToString(t).c_str(), TimeToString(trec).c_str());
         return HS_FILE_ERROR;
      }
      
      if (tend && (t > tend) && !past_end_of_last_file) {
         delete[] buf;
         ::close(fd);
         cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': record %jd timestamp %s is after last timestamp %s", s->fFileName.c_str(), (intmax_t)(irec + count), TimeToString(t).c_str(), TimeToString(tend).c_str());
         return HS_FILE_ERROR;
      }
 
      if (t > end_time)
         break;
      
      char* data = buf + 4;
      
      for (int i=0; i<num_var; i++) {
         int si = var_schema_index[i];
         if (si < 0)
            continue;
         
         if (t < last_time[i]) { // protect against duplicate and non-monotonous data
            delete[] buf;
            ::close(fd);
 
            cm_msg(MERROR, "FileHistory::read_data", "Bad history file \'%s\': record %jd timestamp %s is before timestamp %s of variable %d", s->fFileName.c_str(), (intmax_t)(irec + count), TimeToString(t).c_str(), TimeToString(last_time[i]).c_str(), i);
 
            return HS_FILE_ERROR;
         }
         
         double v = 0;
         void* ptr = data + s->fOffsets[si];
         
         int ii = var_index[i];
         assert(ii >= 0);
         assert(ii < s->fVariables[si].n_data);
         
         switch (s->fVariables[si].type) {
         default:
            // unknown data type
            v = 0;
            break;
         case TID_BYTE:
            v = ((unsigned char*)ptr)[ii];
            break;
         case TID_SBYTE:
            v = ((signed char *)ptr)[ii];
            break;
         case TID_CHAR:
            v = ((char*)ptr)[ii];
            break;
         case TID_WORD:
            v = ((unsigned short *)ptr)[ii];
            break;
         case TID_SHORT:
            v = ((signed short *)ptr)[ii];
            break;
         case TID_DWORD:
            v = ((unsigned int *)ptr)[ii];
            break;
         case TID_INT:
            v = ((int *)ptr)[ii];
            break;
         case TID_BOOL:
            v = ((unsigned int *)ptr)[ii];
            break;
         case TID_FLOAT:
            v = ((float*)ptr)[ii];
            break;
         case TID_DOUBLE:
            v = ((double*)ptr)[ii];
            break;
         }
         
         buffer[i]->Add(t, v);
         last_time[i] = t;
      }
      count++;
   }
 
   delete[] buf;
 
   ::close(fd);
 
   if (debug) {
      printf("FileHistory::read_data: file %s map", s->fFileName.c_str());
      for (size_t i=0; i<var_schema_index.size(); i++) {
         printf(" %2d", var_schema_index[i]);
      }
      printf(" read %d rows\n", count);
   }
 
   if (debug)
      printf("FileHistory::read: file %s, schema time %s..%s, read time %s..%s, %d vars, read %d rows\n", s->fFileName.c_str(), TimeToString(s->fTimeFrom).c_str(), TimeToString(s->fTimeTo).c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), num_var, count);
 
   return HS_SUCCESS;
}
 
//    Implementation of the MidasHistoryInterface     //
 
class SchemaHistoryBase: public MidasHistoryInterface
{
protected:
   int fDebug = 0;
   std::string fConnectString;
 
   // writer data
   HsSchemaVector         fWriterSchema;
   std::vector<HsSchema*> fWriterEvents;
 
   // reader data
   HsSchemaVector         fReaderSchema;
 
public:
   SchemaHistoryBase()
   {
      // empty
   }
 
   virtual ~SchemaHistoryBase()
   {
      for (size_t i=0; i<fWriterEvents.size(); i++) {
         //printf("fWriterEvents[%zu] is %p\n", i, fWriterEvents[i]);
         if (fWriterEvents[i]) {
            delete fWriterEvents[i];
            fWriterEvents[i] = NULL;
         }
      }
      fWriterEvents.clear();
   }
 
   virtual int hs_set_debug(int debug)
   {
      int old = fDebug;
      fDebug = debug;
      return old;
   }
 
   virtual int hs_connect(const char* connect_string) = 0;
   virtual int hs_disconnect() = 0;
 
protected:
   //             Schema functions                       //
 
   // load existing schema
   virtual int read_schema(HsSchemaVector* sv, const char* event_name, const time_t timestamp) = 0; // event_name: =NULL means read only event names, =event_name means load tag names for all matching events; timestamp: =0 means read all schema all the way to the beginning of time, =time means read schema in effect at this time and all newer schema
 
   // return a new copy of a schema for writing into history. If schema for this event does not exist, it will be created
   virtual HsSchema* new_event(const char* event_name, time_t timestamp, int ntags, const TAG tags[]) = 0;
 
   // if output file is too big, close it, create a new output file
   virtual HsSchema* maybe_reopen(const char* event_name, time_t timestamp, HsSchema* s) = 0;
 
public:
   //             Functions used by mlogger              //
 
   int hs_define_event(const char* event_name, time_t timestamp, int ntags, const TAG tags[]);
   int hs_write_event(const char* event_name, time_t timestamp, int buffer_size, const char* buffer);
   int hs_flush_buffers();
 
   //             Functions used by mhttpd               //
 
   int hs_clear_cache();
   int hs_get_events(time_t t, std::vector<std::string> *pevents);
   int hs_get_tags(const char* event_name, time_t t, std::vector<TAG> *ptags);
   int hs_get_last_written(time_t timestamp, int num_var, const char* const event_name[], const char* const var_name[], const int var_index[], time_t last_written[]);
   int hs_read_buffer(time_t start_time, time_t end_time,
                      int num_var, const char* const event_name[], const char* const var_name[], const int var_index[],
                      MidasHistoryBufferInterface* buffer[],
                      int hs_status[]);
 
   /*------------------------------------------------------------------*/
 
   class ReadBuffer: public MidasHistoryBufferInterface
   {
   public:
      time_t fFirstTime;
      time_t fLastTime;
      time_t fInterval;
 
      int fNumAdded;
 
      int      fNumAlloc;
      int     *fNumEntries;
      time_t **fTimeBuffer;
      double **fDataBuffer;
 
      time_t   fPrevTime;
 
      ReadBuffer(time_t first_time, time_t last_time, time_t interval) // ctor
      {
         fNumAdded = 0;
 
         fFirstTime = first_time;
         fLastTime = last_time;
         fInterval = interval;
 
         fNumAlloc = 0;
         fNumEntries = NULL;
         fTimeBuffer = NULL;
         fDataBuffer = NULL;
 
         fPrevTime = 0;
      }
 
      ~ReadBuffer() // dtor
      {
      }
 
      void Realloc(int wantalloc)
      {
         if (wantalloc < fNumAlloc - 10)
            return;
 
         int newalloc = fNumAlloc*2;
 
         if (newalloc <= 1000)
            newalloc = wantalloc + 1000;
 
         //printf("wantalloc %d, fNumEntries %d, fNumAlloc %d, newalloc %d\n", wantalloc, *fNumEntries, fNumAlloc, newalloc);
 
         *fTimeBuffer = (time_t*)realloc(*fTimeBuffer, sizeof(time_t)*newalloc);
         assert(*fTimeBuffer);
 
         *fDataBuffer = (double*)realloc(*fDataBuffer, sizeof(double)*newalloc);
         assert(*fDataBuffer);
 
         fNumAlloc = newalloc;
      }
 
      void Add(time_t t, double v)
      {
         if (t < fFirstTime)
            return;
         if (t > fLastTime)
            return;
 
         fNumAdded++;
 
         if ((fPrevTime==0) || (t >= fPrevTime + fInterval)) {
            int pos = *fNumEntries;
 
            Realloc(pos + 1);
 
            (*fTimeBuffer)[pos] = t;
            (*fDataBuffer)[pos] = v;
 
            (*fNumEntries) = pos + 1;
 
            fPrevTime = t;
         }
      }
 
      void Finish()
      {
 
      }
   };
 
   /*------------------------------------------------------------------*/
 
   int hs_read(time_t start_time, time_t end_time, time_t interval,
               int num_var,
               const char* const event_name[], const char* const var_name[], const int var_index[],
               int num_entries[],
               time_t* time_buffer[], double* data_buffer[],
               int st[]);
   /*------------------------------------------------------------------*/
 
 
   int hs_read_binned(time_t start_time, time_t end_time, int num_bins,
                      int num_var, const char* const event_name[], const char* const var_name[], const int var_index[],
                      int num_entries[],
                      int* count_bins[], double* mean_bins[], double* rms_bins[], double* min_bins[], double* max_bins[],
                      time_t* bins_first_time[], double* bins_first_value[],
                      time_t* bins_last_time[], double* bins_last_value[],
                      time_t last_time[], double last_value[],
                      int st[]);
};
 
MidasHistoryBinnedBuffer::MidasHistoryBinnedBuffer(time_t first_time, time_t last_time, int num_bins) // ctor
{
   fNumEntries = 0;
 
   fNumBins = num_bins;
   fFirstTime = first_time;
   fLastTime = last_time;
 
   fSum0 = new double[num_bins];
   fSum1 = new double[num_bins];
   fSum2 = new double[num_bins];
 
   for (int i=0; i<num_bins; i++) {
      fSum0[i] = 0;
      fSum1[i] = 0;
      fSum2[i] = 0;
   }
}
 
MidasHistoryBinnedBuffer::~MidasHistoryBinnedBuffer() // dtor
{
   delete fSum0; fSum0 = NULL;
   delete fSum1; fSum1 = NULL;
   delete fSum2; fSum2 = NULL;
   // poison the pointers
   fCount = NULL;
   fMean = NULL;
   fRms = NULL;
   fMin = NULL;
   fMax = NULL;
   fBinsFirstTime  = NULL;
   fBinsFirstValue = NULL;
   fBinsLastTime   = NULL;
   fBinsLastValue  = NULL;
   fLastTimePtr    = NULL;
   fLastValuePtr   = NULL;
}
 
void MidasHistoryBinnedBuffer::Start()
{
   for (int ibin = 0; ibin < fNumBins; ibin++) {
      if (fMin)
         fMin[ibin] = 0;
      if (fMax)
         fMax[ibin] = 0;
      if (fBinsFirstTime)
         fBinsFirstTime[ibin] = 0;
      if (fBinsFirstValue)
         fBinsFirstValue[ibin] = 0;
      if (fBinsLastTime)
         fBinsLastTime[ibin] = 0;
      if (fBinsLastValue)
         fBinsLastValue[ibin] = 0;
   }
   if (fLastTimePtr)
      *fLastTimePtr = 0;
   if (fLastValuePtr)
      *fLastValuePtr = 0;
}
 
void MidasHistoryBinnedBuffer::Add(time_t t, double v)
{
   if (t < fFirstTime)
      return;
   if (t > fLastTime)
      return;
 
   fNumEntries++;
 
   double a = (double)(t - fFirstTime);
   double b = (double)(fLastTime - fFirstTime);
   double fbin = fNumBins*a/b;
 
   int ibin = (int)fbin;
 
   if (ibin < 0)
      ibin = 0;
   else if (ibin >= fNumBins)
      ibin = fNumBins-1;
 
   if (fSum0[ibin] == 0) {
      if (fMin)
         fMin[ibin] = v;
      if (fMax)
         fMax[ibin] = v;
      if (fBinsFirstTime)
         fBinsFirstTime[ibin] = t;
      if (fBinsFirstValue)
         fBinsFirstValue[ibin] = v;
      if (fBinsLastTime)
         fBinsLastTime[ibin] = t;
      if (fBinsLastValue)
         fBinsLastValue[ibin] = v;
      if (fLastTimePtr)
         *fLastTimePtr = t;
      if (fLastValuePtr)
         *fLastValuePtr = v;
   }
 
   fSum0[ibin] += 1.0;
   fSum1[ibin] += v;
   fSum2[ibin] += v*v;
 
   if (fMin)
      if (v < fMin[ibin])
         fMin[ibin] = v;
 
   if (fMax)
      if (v > fMax[ibin])
         fMax[ibin] = v;
 
   // NOTE: this assumes t and v are sorted by time.
   if (fBinsLastTime)
      fBinsLastTime[ibin] = t;
   if (fBinsLastValue)
      fBinsLastValue[ibin] = v;
 
   if (fLastTimePtr)
      if (t > *fLastTimePtr) {
         *fLastTimePtr = t;
         if (fLastValuePtr)
            *fLastValuePtr = v;
      }
}
 
void MidasHistoryBinnedBuffer::Finish()
{
   for (int i=0; i<fNumBins; i++) {
      double num = fSum0[i];
      double mean = 0;
      double variance = 0;
      if (num > 0) {
         mean = fSum1[i]/num;
         variance = fSum2[i]/num-mean*mean;
      }
      double rms = 0;
      if (variance > 0)
         rms = sqrt(variance);
 
      if (fCount)
         fCount[i] = (int)num;
 
      if (fMean)
         fMean[i] = mean;
 
      if (fRms)
         fRms[i] = rms;
 
      if (num == 0) {
         if (fMin)
            fMin[i] = 0;
         if (fMax)
            fMax[i] = 0;
      }
   }
}
 
int SchemaHistoryBase::hs_define_event(const char* event_name, time_t timestamp, int ntags, const TAG tags[])
{
   if (fDebug) {
      printf("hs_define_event: event name [%s] with %d tags\n", event_name, ntags);
      if (fDebug > 1)
         PrintTags(ntags, tags);
   }
 
   // delete all events with the same name
   for (size_t i=0; i<fWriterEvents.size(); i++)
      if (fWriterEvents[i])
         if (event_name_cmp(fWriterEvents[i]->fEventName, event_name)==0) {
            if (fDebug)
               printf("deleting exising event %s\n", event_name);
            fWriterEvents[i]->close();
            delete fWriterEvents[i];
            fWriterEvents[i] = NULL;
         }
 
   // check for wrong types etc
   for (int i=0; i<ntags; i++) {
      if (strlen(tags[i].name) < 1) {
         cm_msg(MERROR, "hs_define_event", "Error: History event \'%s\' has empty name at index %d", event_name, i);
         return HS_FILE_ERROR;
      }
      if (tags[i].name[0] == ' ') {
         cm_msg(MERROR, "hs_define_event", "Error: History event \'%s\' has name \'%s\' starting with a blank", event_name, tags[i].name);
         return HS_FILE_ERROR;
      }
      if (tags[i].type <= 0 || tags[i].type >= TID_LAST) {
         cm_msg(MERROR, "hs_define_event", "Error: History event \'%s\' tag \'%s\' at index %d has invalid type %d",
                event_name, tags[i].name, i, tags[i].type);
         return HS_FILE_ERROR;
      }
      if (tags[i].type == TID_STRING) {
         cm_msg(MERROR, "hs_define_event",
                "Error: History event \'%s\' tag \'%s\' at index %d has forbidden type TID_STRING", event_name,
                tags[i].name, i);
         return HS_FILE_ERROR;
      }
      if (tags[i].n_data <= 0) {
         cm_msg(MERROR, "hs_define_event", "Error: History event \'%s\' tag \'%s\' at index %d has invalid n_data %d",
                event_name, tags[i].name, i, tags[i].n_data);
         return HS_FILE_ERROR;
      }
   }
 
   // check for duplicate names. Done by sorting, since this takes only O(N*log*N))
   std::vector<std::string> names;
   for (int i=0; i<ntags; i++) {
      std::string str(tags[i].name);
      std::transform(str.begin(), str.end(), str.begin(), ::toupper);
      names.push_back(str);
   }
   std::sort(names.begin(), names.end());
   for (int i=0; i<ntags-1; i++) {
      if (names[i] == names[i + 1]) {
         cm_msg(MERROR, "hs_define_event",
                "Error: History event \'%s\' has duplicate tag name \'%s\'", event_name,
                names[i].c_str());
         return HS_FILE_ERROR;
      }
   }
 
   HsSchema* s = new_event(event_name, timestamp, ntags, tags);
   if (!s)
      return HS_FILE_ERROR;
 
   s->fDisabled = false;
 
   s->remove_inactive_columns();
 
   // find empty slot in events list
   for (size_t i=0; i<fWriterEvents.size(); i++)
      if (!fWriterEvents[i]) {
         fWriterEvents[i] = s;
         s = NULL;
         break;
      }
 
   // if no empty slots, add at the end
   if (s)
      fWriterEvents.push_back(s);
 
   return HS_SUCCESS;
}
 
int SchemaHistoryBase::hs_write_event(const char* event_name, time_t timestamp, int xbuffer_size, const char* buffer)
{
   if (fDebug)
      printf("hs_write_event: write event \'%s\', time %d, size %d\n", event_name, (int)timestamp, xbuffer_size);
 
   assert(xbuffer_size > 0);
 
   size_t buffer_size = xbuffer_size;
 
   HsSchema *s = NULL;
 
   // find this event
   for (size_t i=0; i<fWriterEvents.size(); i++)
      if (fWriterEvents[i] && (event_name_cmp(fWriterEvents[i]->fEventName, event_name)==0)) {
         s = fWriterEvents[i];
         break;
      }
 
   // not found
   if (!s)
      return HS_UNDEFINED_EVENT;
 
   // deactivated because of error?
   if (s->fDisabled) {
      //printf("HERE!\n");
      return HS_FILE_ERROR;
   }
 
   s = maybe_reopen(event_name, timestamp, s);
 
   assert(s != NULL);
 
   if (s->fNumBytes == 0) { // compute expected data size
      // NB: history data does not have any padding!
      for (size_t i=0; i<s->fVariables.size(); i++) {
         s->fNumBytes += s->fVariables[i].n_bytes;
      }
   }
 
   int status;
 
   if (buffer_size > s->fNumBytes) { // too many bytes!
      if (s->fCountWriteOversize == 0) {
         // only report first occurence
         // count of all occurences is reported by HsSchema destructor
         cm_msg(MERROR, "hs_write_event", "Event \'%s\' data size mismatch: expected %zu bytes, got %zu bytes", s->fEventName.c_str(), s->fNumBytes, buffer_size);
      }
      s->fCountWriteOversize++;
      if (buffer_size > s->fWriteMaxSize)
         s->fWriteMaxSize = buffer_size;
      status = s->write_event(timestamp, buffer, s->fNumBytes);
   } else if (buffer_size < s->fNumBytes) { // too few bytes
      if (s->fCountWriteUndersize == 0) {
         // only report first occurence
         // count of all occurences is reported by HsSchema destructor
         cm_msg(MERROR, "hs_write_event", "Event \'%s\' data size mismatch: expected %zu bytes, got %zu bytes", s->fEventName.c_str(), s->fNumBytes, buffer_size);
      }
      s->fCountWriteUndersize++;
      if (s->fWriteMinSize == 0)
         s->fWriteMinSize = buffer_size;
      else if (buffer_size < s->fWriteMinSize)
         s->fWriteMinSize = buffer_size;
      char* tmp = (char*)malloc(s->fNumBytes);
      memcpy(tmp, buffer, buffer_size);
      memset(tmp + buffer_size, 0, s->fNumBytes - buffer_size);
      status = s->write_event(timestamp, tmp, s->fNumBytes);
      free(tmp);
   } else {
      assert(buffer_size == s->fNumBytes); // obviously
      status = s->write_event(timestamp, buffer, buffer_size);
   }
 
   // if could not write event, deactivate it
   if (status != HS_SUCCESS) {
      s->fDisabled = true;
      cm_msg(MERROR, "hs_write_event", "Event \'%s\' disabled after write error %d", event_name, status);
      return HS_FILE_ERROR;
   }
 
   return HS_SUCCESS;
}
 
int SchemaHistoryBase::hs_flush_buffers()
{
   int status = HS_SUCCESS;
 
   if (fDebug)
      printf("hs_flush_buffers!\n");
 
   for (size_t i=0; i<fWriterEvents.size(); i++)
      if (fWriterEvents[i]) {
         int xstatus = fWriterEvents[i]->flush_buffers();
         if (xstatus != HS_SUCCESS)
            status = xstatus;
      }
 
   return status;
}
 
//             Functions used by mhttpd               //
 
int SchemaHistoryBase::hs_clear_cache()
{
   if (fDebug)
      printf("SchemaHistoryBase::hs_clear_cache!\n");
 
   fWriterSchema.clear();
   fReaderSchema.clear();
 
   return HS_SUCCESS;
}
 
int SchemaHistoryBase::hs_get_events(time_t t, std::vector<std::string> *pevents)
{
   if (fDebug)
      printf("hs_get_events, time %s\n", TimeToString(t).c_str());
 
   int status = read_schema(&fReaderSchema, NULL, t);
   if (status != HS_SUCCESS)
      return status;
 
   if (fDebug) {
      printf("hs_get_events: available schema:\n");
      fReaderSchema.print(false);
   }
 
   assert(pevents);
 
   for (size_t i=0; i<fReaderSchema.size(); i++) {
      HsSchema* s = fReaderSchema[i];
      if (t && s->fTimeTo && s->fTimeTo < t)
         continue;
      bool dupe = false;
      for (size_t j=0; j<pevents->size(); j++)
         if (event_name_cmp((*pevents)[j], s->fEventName.c_str())==0) {
            dupe = true;
            break;
         }
 
      if (!dupe)
         pevents->push_back(s->fEventName);
   }
 
   std::sort(pevents->begin(), pevents->end());
 
   if (fDebug) {
      printf("hs_get_events: returning %zu events\n", pevents->size());
      for (size_t i=0; i<pevents->size(); i++) {
         printf("  %zu: [%s]\n", i, (*pevents)[i].c_str());
      }
   }
 
   return HS_SUCCESS;
}
 
int SchemaHistoryBase::hs_get_tags(const char* event_name, time_t t, std::vector<TAG> *ptags)
{
   if (fDebug)
      printf("hs_get_tags: event [%s], time %s\n", event_name, TimeToString(t).c_str());
 
   assert(ptags);
 
   int status = read_schema(&fReaderSchema, event_name, t);
   if (status != HS_SUCCESS)
      return status;
 
   bool found_event = false;
   for (size_t i=0; i<fReaderSchema.size(); i++) {
      HsSchema* s = fReaderSchema[i];
      if (t && s->fTimeTo && s->fTimeTo < t)
         continue;
 
      if (event_name_cmp(s->fEventName, event_name) != 0)
         continue;
 
      found_event = true;
 
      for (size_t i=0; i<s->fVariables.size(); i++) {
         const char* tagname = s->fVariables[i].name.c_str();
         //printf("event_name [%s], table_name [%s], column name [%s], tag name [%s]\n", event_name, tn.c_str(), cn.c_str(), tagname);
 
         bool dupe = false;
 
         for (size_t k=0; k<ptags->size(); k++)
            if (strcasecmp((*ptags)[k].name, tagname) == 0) {
               dupe = true;
               break;
            }
 
         if (!dupe) {
            TAG t;
            mstrlcpy(t.name, tagname, sizeof(t.name));
            t.type = s->fVariables[i].type;
            t.n_data = s->fVariables[i].n_data;
 
            ptags->push_back(t);
         }
      }
   }
 
   if (!found_event)
      return HS_UNDEFINED_EVENT;
 
   if (fDebug) {
      printf("hs_get_tags: event [%s], returning %zu tags\n", event_name, ptags->size());
      for (size_t i=0; i<ptags->size(); i++) {
         printf("  tag[%zu]: %s[%d] type %d\n", i, (*ptags)[i].name, (*ptags)[i].n_data, (*ptags)[i].type);
      }
   }
 
   return HS_SUCCESS;
}
 
int SchemaHistoryBase::hs_get_last_written(time_t timestamp, int num_var, const char* const event_name[], const char* const var_name[], const int var_index[], time_t last_written[])
{
   if (fDebug) {
      printf("hs_get_last_written: timestamp %s, num_var %d\n", TimeToString(timestamp).c_str(), num_var);
   }
 
   for (int j=0; j<num_var; j++) {
      last_written[j] = 0;
   }
 
   for (int i=0; i<num_var; i++) {
      int status = read_schema(&fReaderSchema, event_name[i], 0);
      if (status != HS_SUCCESS)
         return status;
   }
 
   //fReaderSchema.print(false);
 
   for (int i=0; i<num_var; i++) {
      for (size_t ss=0; ss<fReaderSchema.size(); ss++) {
         HsSchema* s = fReaderSchema[ss];
         // schema is too new
         if (s->fTimeFrom && s->fTimeFrom >= timestamp)
            continue;
         // this schema is newer than last_written and may contain newer data?
         if (s->fTimeFrom && s->fTimeFrom < last_written[i])
            continue;
         // schema for the variables we want?
         int sindex = s->match_event_var(event_name[i], var_name[i], var_index[i]);
         if (sindex < 0)
            continue;
 
         time_t lw = 0;
 
         int status = s->read_last_written(timestamp, fDebug, &lw);
 
         if (status == HS_SUCCESS && lw != 0) {
            for (int j=0; j<num_var; j++) {
               int sj = s->match_event_var(event_name[j], var_name[j], var_index[j]);
               if (sj < 0)
                  continue;
 
               if (lw > last_written[j])
                  last_written[j] = lw;
            }
         }
      }
   }
 
   if (fDebug) {
      printf("hs_get_last_written: timestamp time %s, num_var %d, result:\n", TimeToString(timestamp).c_str(), num_var);
      for (int i=0; i<num_var; i++) {
         printf("  event [%s] tag [%s] index [%d] last_written %s\n", event_name[i], var_name[i], var_index[i], TimeToString(last_written[i]).c_str());
      }
   }
 
   return HS_SUCCESS;
}
 
int SchemaHistoryBase::hs_read_buffer(time_t start_time, time_t end_time,
                                      int num_var, const char* const event_name[], const char* const var_name[], const int var_index[],
                                      MidasHistoryBufferInterface* buffer[],
                                      int hs_status[])
{
   if (fDebug)
      printf("hs_read_buffer: %d variables, start time %s, end time %s\n", num_var, TimeToString(start_time).c_str(), TimeToString(end_time).c_str());
 
   for (int i=0; i<num_var; i++) {
      int status = read_schema(&fReaderSchema, event_name[i], start_time);
      if (status != HS_SUCCESS)
         return status;
   }
 
#if 0
   if (fDebug)
      fReaderSchema.print(false);
#endif
   
   for (int i=0; i<num_var; i++) {
      hs_status[i] = HS_UNDEFINED_VAR;
   }
 
   //for (size_t ss=0; ss<fReaderSchema.size(); ss++) {
   //   HsSchema* s = fReaderSchema[ss];
   //   HsFileSchema* fs = dynamic_cast<HsFileSchema*>(s);
   //   assert(fs != NULL);
   //   printf("schema %d from %s to %s, filename %s\n", ss, TimeToString(fs->fTimeFrom).c_str(),  TimeToString(fs->fTimeTo).c_str(), fs->fFileName.c_str());
   //}
 
   // check that schema are sorted by time
 
#if 0
   // check that schema list is sorted by time, descending fTimeFrom, newest schema first
   for (size_t ss=0; ss<fReaderSchema.size(); ss++) {
      if (fDebug) {
         //printf("Check schema %zu/%zu: prev from %s, this from %s to %s, compare %d %d %d\n", ss, fReaderSchema.size(),
         //       TimeToString(fReaderSchema[ss-1]->fTimeFrom).c_str(),
         //       TimeToString(fReaderSchema[ss]->fTimeFrom).c_str(),
         //       TimeToString(fReaderSchema[ss]->fTimeTo).c_str(),
         //       fReaderSchema[ss-1]->fTimeFrom >= fReaderSchema[ss]->fTimeTo,
         //       fReaderSchema[ss-1]->fTimeFrom > fReaderSchema[ss]->fTimeFrom,
         //       (fReaderSchema[ss-1]->fTimeFrom >= fReaderSchema[ss]->fTimeTo) && (fReaderSchema[ss-1]->fTimeFrom > fReaderSchema[ss]->fTimeFrom));
         printf("Schema %zu/%zu: from %s to %s, name %s\n", ss, fReaderSchema.size(),
                TimeToString(fReaderSchema[ss]->fTimeFrom).c_str(),
                TimeToString(fReaderSchema[ss]->fTimeTo).c_str(),
                fReaderSchema[ss]->fEventName.c_str());
      }
 
      if (ss > 0) {
         //if ((fReaderSchema[ss-1]->fTimeFrom >= fReaderSchema[ss]->fTimeTo) && (fReaderSchema[ss-1]->fTimeFrom > fReaderSchema[ss]->fTimeFrom)) {
         if ((fReaderSchema[ss-1]->fTimeFrom >= fReaderSchema[ss]->fTimeFrom)) {
            // good
         } else {
            cm_msg(MERROR, "SchemaHistoryBase::hs_read_buffer", "History internal error, schema is not ordered by time. Please report this error to the midas forum.");
            return HS_FILE_ERROR;
         }
      }
   }
#endif
 
   std::vector<HsSchema*> slist;
   std::vector<std::vector<int>> smap;
 
   for (size_t ss=0; ss<fReaderSchema.size(); ss++) {
      HsSchema* s = fReaderSchema[ss];
      // schema is too new?
      if (s->fTimeFrom && s->fTimeFrom > end_time)
         continue;
      // schema is too old
      if (s->fTimeTo && s->fTimeTo < start_time)
         continue;
 
      std::vector<int> sm;
 
      for (int i=0; i<num_var; i++) {
         // schema for the variables we want?
         int sindex = s->match_event_var(event_name[i], var_name[i], var_index[i]);
         if (sindex < 0)
            continue;
 
         if (sm.empty()) {
            for (int i=0; i<num_var; i++) {
               sm.push_back(-1);
            }
         }
 
         sm[i] = sindex;
      }
 
      if (!sm.empty()) {
         slist.push_back(s);
         smap.push_back(sm);
      }
   }
 
   if (0||fDebug) {
      printf("Found %zu matching schema:\n", slist.size());
 
      for (size_t i=0; i<slist.size(); i++) {
         HsSchema* s = slist[i];
         s->print();
         for (int k=0; k<num_var; k++)
            printf("  tag %s[%d] sindex %d\n", var_name[k], var_index[k], smap[i][k]);
      }
   }
 
   //for (size_t ss=0; ss<slist.size(); ss++) {
   //   HsSchema* s = slist[ss];
   //   HsFileSchema* fs = dynamic_cast<HsFileSchema*>(s);
   //   assert(fs != NULL);
   //   printf("schema %zu from %s to %s, filename %s", ss, TimeToString(fs->fTimeFrom).c_str(),  TimeToString(fs->fTimeTo).c_str(), fs->fFileName.c_str());
   //   printf(" smap ");
   //   for (int k=0; k<num_var; k++)
   //      printf(" %2d", smap[ss][k]);
   //   printf("\n");
   //}
 
   for (size_t ss=1; ss<slist.size(); ss++) {
      if (fDebug) {
         printf("Check schema %zu/%zu: prev from %s, this from %s to %s, compare %d\n", ss, slist.size(),
                TimeToString(slist[ss-1]->fTimeFrom).c_str(),
                TimeToString(slist[ss]->fTimeFrom).c_str(),
                TimeToString(slist[ss]->fTimeTo).c_str(),
                slist[ss-1]->fTimeFrom >= slist[ss]->fTimeFrom);
      }
      if (slist[ss-1]->fTimeFrom >= slist[ss]->fTimeFrom) {
         // good
      } else {
         cm_msg(MERROR, "SchemaHistoryBase::hs_read_buffer", "History internal error, selected schema is not ordered by time. Please report this error to the midas forum.");
         return HS_FILE_ERROR;
      }
   }
 
   std::vector<time_t> last_time;
 
   for (int i=0; i<num_var; i++) {
      last_time.push_back(start_time);
   }
 
   if (slist.size() > 0) {
      for (size_t i=slist.size()-1; ; i--) {
         HsSchema* s = slist[i];
 
         int status = s->read_data(start_time, end_time, num_var, smap[i], var_index, fDebug, last_time, buffer);
 
         if (status == HS_SUCCESS) {
            for (int j=0; j<num_var; j++) {
               if (smap[i][j] >= 0)
                  hs_status[j] = HS_SUCCESS;
            }
         }
 
         if (i==0)
            break;
      }
   }
 
   return HS_SUCCESS;
}
 
int SchemaHistoryBase::hs_read(time_t start_time, time_t end_time, time_t interval,
                               int num_var,
                               const char* const event_name[], const char* const var_name[], const int var_index[],
                               int num_entries[],
                               time_t* time_buffer[], double* data_buffer[],
                               int st[])
{
   int status;
 
   ReadBuffer** buffer = new ReadBuffer*[num_var];
   MidasHistoryBufferInterface** bi = new MidasHistoryBufferInterface*[num_var];
 
   for (int i=0; i<num_var; i++) {
      buffer[i] = new ReadBuffer(start_time, end_time, interval);
      bi[i] = buffer[i];
 
      // make sure outputs are initialized to something sane
      if (num_entries)
         num_entries[i] = 0;
      if (time_buffer)
         time_buffer[i] = NULL;
      if (data_buffer)
         data_buffer[i] = NULL;
      if (st)
         st[i] = 0;
 
      if (num_entries)
         buffer[i]->fNumEntries = &num_entries[i];
      if (time_buffer)
         buffer[i]->fTimeBuffer = &time_buffer[i];
      if (data_buffer)
         buffer[i]->fDataBuffer = &data_buffer[i];
   }
 
   status = hs_read_buffer(start_time, end_time,
                           num_var, event_name, var_name, var_index,
                           bi, st);
 
   for (int i=0; i<num_var; i++) {
      buffer[i]->Finish();
      delete buffer[i];
   }
 
   delete[] buffer;
   delete[] bi;
 
   return status;
}
 
int SchemaHistoryBase::hs_read_binned(time_t start_time, time_t end_time, int num_bins,
                                      int num_var, const char* const event_name[], const char* const var_name[], const int var_index[],
                                      int num_entries[],
                                      int* count_bins[], double* mean_bins[], double* rms_bins[], double* min_bins[], double* max_bins[],
                                      time_t* bins_first_time[], double* bins_first_value[],
                                      time_t* bins_last_time[], double* bins_last_value[],
                                      time_t last_time[], double last_value[],
                                      int st[])
{
   int status;
 
   MidasHistoryBinnedBuffer** buffer = new MidasHistoryBinnedBuffer*[num_var];
   MidasHistoryBufferInterface** xbuffer = new MidasHistoryBufferInterface*[num_var];
 
   for (int i=0; i<num_var; i++) {
      buffer[i] = new MidasHistoryBinnedBuffer(start_time, end_time, num_bins);
      xbuffer[i] = buffer[i];
 
      if (count_bins)
         buffer[i]->fCount = count_bins[i];
      if (mean_bins)
         buffer[i]->fMean = mean_bins[i];
      if (rms_bins)
         buffer[i]->fRms = rms_bins[i];
      if (min_bins)
         buffer[i]->fMin = min_bins[i];
      if (max_bins)
         buffer[i]->fMax = max_bins[i];
      if (bins_first_time)
         buffer[i]->fBinsFirstTime = bins_first_time[i];
      if (bins_first_value)
         buffer[i]->fBinsFirstValue = bins_first_value[i];
      if (bins_last_time)
         buffer[i]->fBinsLastTime = bins_last_time[i];
      if (bins_last_value)
         buffer[i]->fBinsLastValue = bins_last_value[i];
      if (last_time)
         buffer[i]->fLastTimePtr = &last_time[i];
      if (last_value)
         buffer[i]->fLastValuePtr = &last_value[i];
 
      buffer[i]->Start();
   }
 
   status = hs_read_buffer(start_time, end_time,
                           num_var, event_name, var_name, var_index,
                           xbuffer,
                           st);
 
   for (int i=0; i<num_var; i++) {
      buffer[i]->Finish();
      if (num_entries)
         num_entries[i] = buffer[i]->fNumEntries;
      if (0) {
         for (int j=0; j<num_bins; j++) {
            printf("var %d bin %d count %d, first %s last %s value first %f last %f\n", i, j, count_bins[i][j], TimeToString(bins_first_time[i][j]).c_str(), TimeToString(bins_last_time[i][j]).c_str(), bins_first_value[i][j], bins_last_value[i][j]);
         }
      }
      delete buffer[i];
   }
 
   delete[] buffer;
   delete[] xbuffer;
 
   return status;
}
 
//                    SQL schema                      //
 
int HsSqlSchema::close_transaction()
{
   if (!fSql->IsConnected()) {
      return HS_SUCCESS;
   }
   
   int status = HS_SUCCESS;
   if (get_transaction_count() > 0) {
      status = fSql->CommitTransaction(fTableName.c_str());
      reset_transaction_count();
   }
   return status;
}
 
int HsSchema::match_event_var(const char* event_name, const char* var_name, const int var_index)
{
   if (!MatchEventName(this->fEventName.c_str(), event_name))
      return -1;
 
   for (size_t j=0; j<this->fVariables.size(); j++) {
      if (MatchTagName(this->fVariables[j].name.c_str(), this->fVariables[j].n_data, var_name, var_index)) {
         // Second clause in if() is case where MatchTagName used the "alternate tag name".
         // E.g. our variable name is "IM05[3]" (n_data=1), but we're looking for var_name="IM05" and var_index=3.
         if (var_index < this->fVariables[j].n_data || (this->fVariables[j].n_data == 1 && this->fVariables[j].name.find("[") != std::string::npos)) {
            return j;
         }
      }
   }
 
   return -1;
}
 
int HsSqlSchema::match_event_var(const char* event_name, const char* var_name, const int var_index)
{
   if (event_name_cmp(this->fTableName, event_name)==0) {
      for (size_t j=0; j<this->fVariables.size(); j++) {
         if (var_name_cmp(this->fColumnNames[j], var_name)==0)
            return j;
      }
   }
 
   return HsSchema::match_event_var(event_name, var_name, var_index);
}
 
static HsSqlSchema* NewSqlSchema(HsSchemaVector* sv, const char* table_name, time_t t)
{
   time_t tt = 0;
   int j=-1;
   int jjx=-1; // remember oldest schema
   time_t ttx = 0;
   for (size_t i=0; i<sv->size(); i++) {
      HsSqlSchema* s = (HsSqlSchema*)(*sv)[i];
      if (s->fTableName != table_name)
         continue;
 
      if (s->fTimeFrom == t) {
         return s;
      }
 
      // remember the last schema before time t
      if (s->fTimeFrom < t) {
         if (s->fTimeFrom > tt) {
            tt = s->fTimeFrom;
            j = i;
         }
      }
 
      if (jjx < 0) {
         jjx = i;
         ttx = s->fTimeFrom;
      }
 
      if (s->fTimeFrom < ttx) {
         jjx = i;
         ttx = s->fTimeFrom;
      }
 
      //printf("table_name [%s], t=%s, i=%d, j=%d %s, tt=%s, dt is %d\n", table_name, TimeToString(t).c_str(), i, j, TimeToString(s->fTimeFrom).c_str(), TimeToString(tt).c_str(), (int)(s->fTimeFrom-t));
   }
 
   //printf("NewSqlSchema: will copy schema j=%d, tt=%d at time %d\n", j, tt, t);
 
   //printf("cloned schema at time %s: ", TimeToString(t).c_str());
   //(*sv)[j]->print(false);
 
   //printf("schema before:\n");
   //sv->print(false);
 
   if (j >= 0) {
      HsSqlSchema* s = new HsSqlSchema;
      *s = *(HsSqlSchema*)(*sv)[j]; // make a copy
      s->fTimeFrom = t;
      sv->add(s);
 
      //printf("schema after:\n");
      //sv->print(false);
 
      return s;
   }
 
   if (jjx >= 0) {
      cm_msg(MERROR, "NewSqlSchema", "Error: Unexpected ordering of schema for table \'%s\', good luck!", table_name);
 
      HsSqlSchema* s = new HsSqlSchema;
      *s = *(HsSqlSchema*)(*sv)[jjx]; // make a copy
      s->fTimeFrom = t;
      s->fTimeTo = ttx;
      sv->add(s);
 
      //printf("schema after:\n");
      //sv->print(false);
 
      return s;
   }
 
   cm_msg(MERROR, "NewSqlSchema", "Error: Cannot clone schema for table \'%s\', good luck!", table_name);
   return NULL;
}
 
void HsSqlSchema::remove_inactive_columns()
{
   assert(fVariables.size() == fColumnInactive.size());
   assert(fVariables.size() == fColumnNames.size());
   assert(fVariables.size() == fColumnTypes.size());
   assert(fVariables.size() == fOffsets.size());
 
   size_t count_active = 0;
   size_t count_inactive = 0;
 
   for (size_t i=0; i<fColumnInactive.size(); i++) {
      if (fColumnInactive[i])
         count_inactive += 1;
      else
         count_active += 1;
   }
 
   //printf("remove_inactive_columns: enter! count_active: %zu, count_inactive: %zu\n", count_active, count_inactive);
   //print();
 
   if (count_inactive > 0) {
      size_t j=0;
 
      for (size_t i=0; i<fColumnInactive.size(); i++) {
         if (fColumnInactive[i]) {
            // skip this entry
         } else {
            if (j != i) {
               fVariables[j]       = fVariables[i];
               fColumnInactive[j] = fColumnInactive[i];
               fColumnNames[j]    = fColumnNames[i];
               fColumnTypes[j]    = fColumnTypes[i];
               fOffsets[j]         = fOffsets[i];
            }
            j++;
         }
      }
 
      //print();
      //printf("%zu %zu\n", j, count_active);
 
      assert(j == count_active);
 
      //print();
 
      fVariables.resize(count_active);
      fColumnInactive.resize(count_active);
      fColumnNames.resize(count_active);
      fColumnTypes.resize(count_active);
      fOffsets.resize(count_active);
 
      assert(fVariables.size() == fColumnInactive.size());
      assert(fVariables.size() == fColumnNames.size());
      assert(fVariables.size() == fColumnTypes.size());
      assert(fVariables.size() == fOffsets.size());
      
      //printf("remove_inactice_columns: exit!\n");
      //print();
   }
}
 
int HsSqlSchema::write_event(const time_t t, const char* data, const size_t data_size)
{
   HsSqlSchema* s = this;
 
   assert(s->fVariables.size() == s->fColumnInactive.size());
   assert(s->fVariables.size() == s->fColumnNames.size());
   assert(s->fVariables.size() == s->fColumnTypes.size());
   assert(s->fVariables.size() == s->fOffsets.size());
 
   std::string tags;
   std::string values;
 
   for (size_t i=0; i<s->fVariables.size(); i++) {
      // NB: inactive columns should have been removed from the schema. K.O.
 
      if (s->fColumnInactive[i]) {
         cm_msg(MERROR, "HsSqlSchema::write_event", "Internal error, unexpected inactive column %zu", i);
         cm_msg_flush_buffer();
         return HS_FILE_ERROR;
      }
 
      int type   = s->fVariables[i].type;
      int n_data = s->fVariables[i].n_data;
      int offset = s->fOffsets[i];
      const char* column_name = s->fColumnNames[i].c_str();
 
      if (offset < 0) {
         cm_msg(MERROR, "HsSqlSchema::write_event", "Internal error, unexpected negative offset %d for column %zu", offset, i);
         cm_msg_flush_buffer();
         return HS_FILE_ERROR;
      }
 
      assert(n_data == 1);
      assert(strlen(column_name) > 0);
      assert(offset >= 0);
      assert((size_t)offset < data_size);
 
      void* ptr = (void*)(data+offset);
 
      tags += ", ";
      tags += fSql->QuoteId(column_name);
 
      values += ", ";
 
      char buf[1024];
      int j=0;
 
      switch (type) {
      default:
         sprintf(buf, "unknownType%d", type);
         break;
      case TID_BYTE:
         sprintf(buf, "%u",((unsigned char *)ptr)[j]);
         break;
      case TID_SBYTE:
         sprintf(buf, "%d",((signed char*)ptr)[j]);
         break;
      case TID_CHAR:
         // FIXME: quotes
         sprintf(buf, "\'%c\'",((char*)ptr)[j]);
         break;
      case TID_WORD:
         sprintf(buf, "%u",((unsigned short *)ptr)[j]);
         break;
      case TID_SHORT:
         sprintf(buf, "%d",((short *)ptr)[j]);
         break;
      case TID_DWORD:
         sprintf(buf, "%u",((unsigned int *)ptr)[j]);
         break;
      case TID_INT:
         sprintf(buf, "%d",((int *)ptr)[j]);
         break;
      case TID_BOOL:
         sprintf(buf, "%u",((unsigned int *)ptr)[j]);
         break;
      case TID_FLOAT:
         // FIXME: quotes
         sprintf(buf, "\'%.8g\'",((float*)ptr)[j]);
         break;
      case TID_DOUBLE:
         // FIXME: quotes
         sprintf(buf, "\'%.16g\'",((double*)ptr)[j]);
         break;
      }
 
      values += buf;
   }
 
   // 2001-02-16 20:38:40.1
   struct tm tms;
   localtime_r(&t, &tms); // somebody must call tzset() before this.
   char buf[1024];
   strftime(buf, sizeof(buf)-1, "%Y-%m-%d %H:%M:%S.0", &tms);
 
   std::string cmd;
   cmd = "INSERT INTO ";
   cmd += fSql->QuoteId(s->fTableName.c_str());
   cmd += " (_t_time, _i_time";
   cmd += tags;
   cmd += ") VALUES (";
   cmd += fSql->QuoteString(buf);
   cmd += ", ";
   cmd += fSql->QuoteString(TimeToString(t).c_str());
   cmd += "";
   cmd += values;
   cmd += ");";
 
   if (fSql->IsConnected()) {
      if (s->get_transaction_count() == 0)
         fSql->OpenTransaction(s->fTableName.c_str());
 
      s->increment_transaction_count();
 
      int status = fSql->Exec(s->fTableName.c_str(), cmd.c_str());
 
      // mh2sql who does not call hs_flush_buffers()
      // so we should flush the transaction by hand
      // some SQL engines have limited transaction buffers... K.O.
      if (s->get_transaction_count() > 100000) {
         //printf("flush table %s\n", table_name);
         fSql->CommitTransaction(s->fTableName.c_str());
         s->reset_transaction_count();
      }
      
      if (status != DB_SUCCESS) {
         return status;
      }
   } else {
      int status = fSql->ExecDisconnected(s->fTableName.c_str(), cmd.c_str());
      if (status != DB_SUCCESS) {
         return status;
      }
   }
 
   return HS_SUCCESS;
}
 
int HsSqlSchema::read_last_written(const time_t timestamp,
                                   const int debug,
                                   time_t* last_written)
{
   if (debug)
      printf("SqlHistory::read_last_written: table [%s], timestamp %s\n", fTableName.c_str(), TimeToString(timestamp).c_str());
 
   std::string cmd;
   cmd += "SELECT _i_time FROM ";
   cmd += fSql->QuoteId(fTableName.c_str());
   cmd += " WHERE _i_time < ";
   cmd += TimeToString(timestamp);
   cmd += " ORDER BY _i_time DESC LIMIT 2;";
 
   int status = fSql->Prepare(fTableName.c_str(), cmd.c_str());
 
   if (status != DB_SUCCESS)
      return status;
 
   time_t lw = 0;
 
   /* Loop through the rows in the result-set */
 
   while (1) {
      status = fSql->Step();
      if (status != DB_SUCCESS)
         break;
 
      time_t t = fSql->GetTime(0);
 
      if (t >= timestamp)
         continue;
 
      if (t > lw)
         lw = t;
   }
 
   fSql->Finalize();
 
   *last_written = lw;
 
   if (debug)
      printf("SqlHistory::read_last_written: table [%s], timestamp %s, last_written %s\n", fTableName.c_str(), TimeToString(timestamp).c_str(), TimeToString(lw).c_str());
 
   return HS_SUCCESS;
}
 
int HsSqlSchema::read_data(const time_t start_time,
                           const time_t end_time,
                           const int num_var, const std::vector<int>& var_schema_index, const int var_index[],
                           const int debug,
                           std::vector<time_t>& last_time,
                           MidasHistoryBufferInterface* buffer[])
{
   bool bad_last_time = false;
 
   if (debug)
      printf("SqlHistory::read_data: table [%s], start %s, end %s\n", fTableName.c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str());
 
   std::string collist;
 
   for (int i=0; i<num_var; i++) {
      int j = var_schema_index[i];
      if (j < 0)
         continue;
      if (collist.length() > 0)
         collist += ", ";
      collist += fSql->QuoteId(fColumnNames[j].c_str());
   }
 
   std::string cmd;
   cmd += "SELECT _i_time, ";
   cmd += collist;
   cmd += " FROM ";
   cmd += fSql->QuoteId(fTableName.c_str());
   cmd += " WHERE _i_time>=";
   cmd += TimeToString(start_time);
   cmd += " and _i_time<=";
   cmd += TimeToString(end_time);
   cmd += " ORDER BY _i_time;";
 
   int status = fSql->Prepare(fTableName.c_str(), cmd.c_str());
 
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   /* Loop through the rows in the result-set */
 
   int count = 0;
 
   while (1) {
      status = fSql->Step();
      if (status != DB_SUCCESS)
         break;
 
      count++;
 
      time_t t = fSql->GetTime(0);
 
      if (t < start_time || t > end_time)
         continue;
 
      int k = 0;
 
      for (int i=0; i<num_var; i++) {
         int j = var_schema_index[i];
         if (j < 0)
            continue;
 
         if (t < last_time[i]) { // protect against duplicate and non-monotonous data
            bad_last_time = true;
         } else {
            double v = fSql->GetDouble(1+k);
 
            //printf("Column %d, index %d, Row %d, time %d, value %f\n", k, colindex[k], count, t, v);
 
            buffer[i]->Add(t, v);
            last_time[i] = t;
         }
 
         k++;
      }
   }
 
   fSql->Finalize();
 
   if (bad_last_time) {
      cm_msg(MERROR, "SqlHistory::read_data", "Detected duplicate or non-monotonous data in table \"%s\" for start time %s and end time %s", fTableName.c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str());
   }
 
   if (debug)
      printf("SqlHistory::read_data: table [%s], start %s, end %s, read %d rows\n", fTableName.c_str(), TimeToString(start_time).c_str(), TimeToString(end_time).c_str(), count);
 
   return HS_SUCCESS;
}
 
int HsSqlSchema::get_transaction_count() {
   if (!fSql || fSql->fTransactionPerTable) {
      return fTableTransactionCount;
   } else {
      return gfTransactionCount[fSql];
   }
}
 
void HsSqlSchema::reset_transaction_count() {
   if (!fSql || fSql->fTransactionPerTable) {
      fTableTransactionCount = 0;
   } else {
      gfTransactionCount[fSql] = 0;
   }
}
 
void HsSqlSchema::increment_transaction_count() {
   if (!fSql || fSql->fTransactionPerTable) {
      fTableTransactionCount++;
   } else {
      gfTransactionCount[fSql]++;
   }
}
 
//             SQL history functions                  //
 
static int StartSqlTransaction(SqlBase* sql, const char* table_name, bool* have_transaction)
{
   if (*have_transaction)
      return HS_SUCCESS;
 
   int status = sql->OpenTransaction(table_name);
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   *have_transaction = true;
   return HS_SUCCESS;
}
 
static int CreateSqlTable(SqlBase* sql, const char* table_name, bool* have_transaction, bool set_default_timestamp = false)
{
   int status;
 
   status = StartSqlTransaction(sql, table_name, have_transaction);
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   std::string cmd;
 
   cmd = "CREATE TABLE ";
   cmd += sql->QuoteId(table_name);
   if (set_default_timestamp) {
      cmd += " (_t_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, _i_time INTEGER NOT NULL DEFAULT 0);";
   } else {
      cmd += " (_t_time TIMESTAMP NOT NULL, _i_time INTEGER NOT NULL);";
   }
 
   status = sql->Exec(table_name, cmd.c_str());
 
 
   if (status == DB_KEY_EXIST) {
      cm_msg(MINFO, "CreateSqlTable", "Adding SQL table \"%s\", but it already exists", table_name);
      cm_msg_flush_buffer();
      return status;
   }
 
   if (status != DB_SUCCESS) {
      cm_msg(MINFO, "CreateSqlTable", "Adding SQL table \"%s\", error status %d", table_name, status);
      cm_msg_flush_buffer();
      return HS_FILE_ERROR;
   }
 
   cm_msg(MINFO, "CreateSqlTable", "Adding SQL table \"%s\"", table_name);
   cm_msg_flush_buffer();
 
   std::string i_index_name;
   i_index_name = table_name;
   i_index_name += "_i_time_index";
 
   std::string t_index_name;
   t_index_name = table_name;
   t_index_name += "_t_time_index";
 
   cmd = "CREATE INDEX ";
   cmd += sql->QuoteId(i_index_name.c_str());
   cmd += " ON ";
   cmd += sql->QuoteId(table_name);
   cmd += " (_i_time ASC);";
 
   status = sql->Exec(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   cmd = "CREATE INDEX ";
   cmd += sql->QuoteId(t_index_name.c_str());
   cmd += " ON ";
   cmd += sql->QuoteId(table_name);
   cmd += " (_t_time);";
 
   status = sql->Exec(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   return status;
}
 
static int CreateSqlHyperTable(SqlBase* sql, const char* table_name, bool* have_transaction) {
   int status;
 
   status = StartSqlTransaction(sql, table_name, have_transaction);
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   std::string cmd;
 
   cmd = "CREATE TABLE ";
   cmd += sql->QuoteId(table_name);
   cmd += " (_t_time TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, _i_time INTEGER NOT NULL DEFAULT 0);";
 
   status = sql->Exec(table_name, cmd.c_str());
 
   if (status == DB_KEY_EXIST) {
      cm_msg(MINFO, "CreateSqlHyperTable", "Adding SQL table \"%s\", but it already exists", table_name);
      cm_msg_flush_buffer();
      return status;
   }
 
   if (status != DB_SUCCESS) {
      cm_msg(MINFO, "CreateSqlHyperTable", "Adding SQL table \"%s\", error status %d", table_name, status);
      cm_msg_flush_buffer();
      return HS_FILE_ERROR;
   }
 
   cm_msg(MINFO, "CreateSqlHyperTable", "Adding SQL table \"%s\"", table_name);
   cm_msg_flush_buffer();
 
   cmd = "SELECT create_hypertable(";
   cmd += sql->QuoteString(table_name);
   cmd += ", '_t_time');";
   
   // convert regular table to hypertable
   status = sql->Exec(table_name, cmd.c_str());
 
   if (status != DB_SUCCESS) {
      cm_msg(MINFO, "CreateSqlHyperTable", "Converting SQL table to hypertable \"%s\", error status %d", table_name, status);
      cm_msg_flush_buffer();
      return HS_FILE_ERROR;
   }
 
   std::string i_index_name;
   i_index_name = table_name;
   i_index_name += "_i_time_index";
 
   std::string t_index_name;
   t_index_name = table_name;
   t_index_name += "_t_time_index";
 
   cmd = "CREATE INDEX ";
   cmd += sql->QuoteId(i_index_name.c_str());
   cmd += " ON ";
   cmd += sql->QuoteId(table_name);
   cmd += " (_i_time ASC);";
 
   status = sql->Exec(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   cmd = "CREATE INDEX ";
   cmd += sql->QuoteId(t_index_name.c_str());
   cmd += " ON ";
   cmd += sql->QuoteId(table_name);
   cmd += " (_t_time);";
 
   status = sql->Exec(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   return status;
}
 
static int CreateSqlColumn(SqlBase* sql, const char* table_name, const char* column_name, const char* column_type, bool* have_transaction, int debug)
{
   if (debug)
      printf("CreateSqlColumn: table [%s], column [%s], type [%s]\n", table_name, column_name, column_type);
 
   int status = StartSqlTransaction(sql, table_name, have_transaction);
   if (status != HS_SUCCESS)
      return status;
 
   std::string cmd;
   cmd = "ALTER TABLE ";
   cmd += sql->QuoteId(table_name);
   cmd += " ADD COLUMN ";
   cmd += sql->QuoteId(column_name);
   cmd += " ";
   cmd += column_type;
   cmd += ";";
 
   status = sql->Exec(table_name, cmd.c_str());
 
   cm_msg(MINFO, "CreateSqlColumn", "Adding column \"%s\" to SQL table \"%s\", status %d", column_name, table_name, status);
   cm_msg_flush_buffer();
 
   return status;
}
 
//             SQL history base classes               //
 
class SqlHistoryBase: public SchemaHistoryBase
{
public:
   SqlBase *fSql;
 
   SqlHistoryBase() // ctor
   {
      fSql = NULL;
      hs_clear_cache();
   }
 
   virtual ~SqlHistoryBase() // dtor
   {
      hs_disconnect();
      if (fSql)
         delete fSql;
      fSql = NULL;
   }
 
   int hs_set_debug(int debug)
   {
      if (fSql)
         fSql->fDebug = debug;
      return SchemaHistoryBase::hs_set_debug(debug);
   }
 
   int hs_connect(const char* connect_string);
   int hs_disconnect();
   HsSchema* new_event(const char* event_name, time_t timestamp, int ntags, const TAG tags[]);
   int read_schema(HsSchemaVector* sv, const char* event_name, const time_t timestamp);
   HsSchema* maybe_reopen(const char* event_name, time_t timestamp, HsSchema* s) { return s; };
 
 
protected:
   virtual int read_table_and_event_names(HsSchemaVector *sv) = 0;
   virtual int read_column_names(HsSchemaVector *sv, const char* table_name, const char* event_name) = 0;
   virtual int create_table(HsSchemaVector* sv, const char* event_name, time_t timestamp) = 0;
   virtual int update_column(const char* event_name, const char* table_name, const char* column_name, const char* column_type, const char* tag_name, const char* tag_type, const time_t timestamp, bool active, bool* have_transaction) = 0;
 
   int update_schema(HsSqlSchema* s, const time_t timestamp, const int ntags, const TAG tags[], bool write_enable);
   int update_schema1(HsSqlSchema* s, const time_t timestamp, const int ntags, const TAG tags[], bool write_enable, bool* have_transaction);
};
 
int SqlHistoryBase::hs_connect(const char* connect_string)
{
   if (fDebug)
      printf("hs_connect [%s]!\n", connect_string);
 
   assert(fSql);
 
   if (fSql->IsConnected())
      if (strcmp(fConnectString.c_str(), connect_string) == 0)
         return HS_SUCCESS;
 
   hs_disconnect();
 
   if (!connect_string || strlen(connect_string) < 1) {
      // FIXME: should use "logger dir" or some such default, that code should be in hs_get_history(), not here
      connect_string = ".";
   }
 
   fConnectString = connect_string;
 
   if (fDebug)
      printf("hs_connect: connecting to SQL database \'%s\'\n", fConnectString.c_str());
 
   int status = fSql->Connect(fConnectString.c_str());
   if (status != DB_SUCCESS)
      return status;
 
   return HS_SUCCESS;
}
 
int SqlHistoryBase::hs_disconnect()
{
   if (fDebug)
      printf("hs_disconnect!\n");
 
   hs_flush_buffers();
 
   fSql->Disconnect();
 
   hs_clear_cache();
 
   return HS_SUCCESS;
}
 
HsSchema* SqlHistoryBase::new_event(const char* event_name, time_t timestamp, int ntags, const TAG tags[])
{
   if (fDebug)
      printf("SqlHistory::new_event: event [%s], timestamp %s, ntags %d\n", event_name, TimeToString(timestamp).c_str(), ntags);
 
   int status;
 
   if (fWriterSchema.size() == 0) {
      status = read_table_and_event_names(&fWriterSchema);
      if (status != HS_SUCCESS)
         return NULL;
   }
 
   HsSqlSchema* s = (HsSqlSchema*)fWriterSchema.find_event(event_name, timestamp);
 
   // schema does not exist, the SQL tables probably do not exist yet
 
   if (!s) {
      status = create_table(&fWriterSchema, event_name, timestamp);
      if (status != HS_SUCCESS)
         return NULL;
 
      s = (HsSqlSchema*)fWriterSchema.find_event(event_name, timestamp);
 
      if (!s) {
         cm_msg(MERROR, "SqlHistory::new_event", "Error: Cannot create schema for event \'%s\', see previous messages", event_name);
         fWriterSchema.find_event(event_name, timestamp, 1);
         return NULL;
      }
   }
 
   assert(s != NULL);
 
   status = read_column_names(&fWriterSchema, s->fTableName.c_str(), s->fEventName.c_str());
   if (status != HS_SUCCESS)
      return NULL;
 
   s = (HsSqlSchema*)fWriterSchema.find_event(event_name, timestamp);
 
   if (!s) {
      cm_msg(MERROR, "SqlHistory::new_event", "Error: Cannot update schema database for event \'%s\', see previous messages", event_name);
      return NULL;
   }
 
   if (0||fDebug) {
      printf("SqlHistory::new_event: schema for [%s] is %p\n", event_name, s);
      if (s)
         s->print();
   }
 
   status = update_schema(s, timestamp, ntags, tags, true);
   if (status != HS_SUCCESS) {
      cm_msg(MERROR, "SqlHistory::new_event", "Error: Cannot create schema for event \'%s\', see previous messages", event_name);
      return NULL;
   }
 
   status = read_column_names(&fWriterSchema, s->fTableName.c_str(), s->fEventName.c_str());
   if (status != HS_SUCCESS)
      return NULL;
 
   s = (HsSqlSchema*)fWriterSchema.find_event(event_name, timestamp);
 
   if (!s) {
      cm_msg(MERROR, "SqlHistory::new_event", "Error: Cannot update schema database for event \'%s\', see previous messages", event_name);
      return NULL;
   }
 
   if (0||fDebug) {
      printf("SqlHistory::new_event: schema for [%s] is %p\n", event_name, s);
      if (s)
         s->print();
   }
 
   // last call to UpdateMysqlSchema with "false" will check that new schema matches the new tags
 
   status = update_schema(s, timestamp, ntags, tags, false);
   if (status != HS_SUCCESS) {
      cm_msg(MERROR, "SqlHistory::new_event", "Error: Cannot create schema for event \'%s\', see previous messages", event_name);
      //fDebug = 1;
      //update_schema(s, timestamp, ntags, tags, false);
      //abort();
      return NULL;
   }
 
   HsSqlSchema* e = new HsSqlSchema();
 
   *e = *s; // make a copy of the schema
 
   return e;
}
 
int SqlHistoryBase::read_schema(HsSchemaVector* sv, const char* event_name, const time_t timestamp)
{
   if (fDebug)
      printf("SqlHistory::read_schema: loading schema for event [%s] at time %s\n", event_name, TimeToString(timestamp).c_str());
 
   int status;
 
   if (sv->size() == 0) {
      status = read_table_and_event_names(sv);
      if (status != HS_SUCCESS)
         return status;
   }
 
   //sv->print(false);
 
   if (event_name == NULL)
      return HS_SUCCESS;
 
   for (size_t i=0; i<sv->size(); i++) {
      HsSqlSchema* h = (HsSqlSchema*)(*sv)[i];
      // skip schema with already read column names
      if (h->fVariables.size() > 0)
         continue;
      // skip schema with different name
      if (!MatchEventName(h->fEventName.c_str(), event_name))
         continue;
 
      size_t nn = sv->size();
 
      status = read_column_names(sv, h->fTableName.c_str(), h->fEventName.c_str());
 
      // if new schema was added, loop all over again
      if (sv->size() != nn)
         i=0;
   }
 
   //sv->print(false);
 
   return HS_SUCCESS;
}
 
int SqlHistoryBase::update_schema(HsSqlSchema* s, const time_t timestamp, const int ntags, const TAG tags[], bool write_enable)
{
   int status;
   bool have_transaction = false;
 
   status = update_schema1(s, timestamp, ntags, tags, write_enable, &have_transaction);
 
   if (have_transaction) {
      int xstatus;
 
      if (status == HS_SUCCESS)
         xstatus = fSql->CommitTransaction(s->fTableName.c_str());
      else
         xstatus = fSql->RollbackTransaction(s->fTableName.c_str());
 
      if (xstatus != DB_SUCCESS) {
         return HS_FILE_ERROR;
      }
      have_transaction = false;
   }
 
   return status;
}
 
int SqlHistoryBase::update_schema1(HsSqlSchema* s, const time_t timestamp, const int ntags, const TAG tags[], bool write_enable, bool* have_transaction)
{
   int status;
 
   if (fDebug)
      printf("update_schema1\n");
 
   // check that compare schema with tags[]
 
   bool schema_ok = true;
 
   int offset = 0;
   for (int i=0; i<ntags; i++) {
      for (unsigned int j=0; j<tags[i].n_data; j++) {
         int         tagtype = tags[i].type;
         std::string tagname = tags[i].name;
         std::string maybe_colname = MidasNameToSqlName(tags[i].name);
 
         if (tags[i].n_data > 1) {
            char s[256];
            sprintf(s, "[%d]", j);
            tagname += s;
 
            sprintf(s, "_%d", j);
            maybe_colname += s;
         }
 
         int count = 0;
 
         for (size_t j=0; j<s->fVariables.size(); j++) {
            // NB: inactive columns will be reactivated or recreated by the if(count==0) branch. K.O.
            if (s->fColumnInactive[j])
               continue;
            if (tagname == s->fVariables[j].name) {
               if (s->fSql->TypesCompatible(tagtype, s->fColumnTypes[j].c_str())) {
                  if (count == 0) {
                     s->fOffsets[j] = offset;
                     offset += rpc_tid_size(tagtype);
                  }
                  count++;
                  if (count > 1) {
                     cm_msg(MERROR, "SqlHistory::update_schema", "Duplicate SQL column \'%s\' type \'%s\' in table \"%s\" with MIDAS type \'%s\' history event \"%s\" tag \"%s\"", s->fColumnNames[j].c_str(), s->fColumnTypes[j].c_str(), s->fTableName.c_str(), rpc_tid_name(tagtype), s->fEventName.c_str(), tagname.c_str());
                     cm_msg_flush_buffer();
                  }
               } else {
                  // column with incompatible type, mark it as unused
                  schema_ok = false;
                  if (fDebug)
                     printf("Incompatible column!\n");
                  if (write_enable) {
                     cm_msg(MINFO, "SqlHistory::update_schema", "Deactivating SQL column \'%s\' type \'%s\' in table \"%s\" as incompatible with MIDAS type \'%s\' history event \"%s\" tag \"%s\"", s->fColumnNames[j].c_str(), s->fColumnTypes[j].c_str(), s->fTableName.c_str(), rpc_tid_name(tagtype), s->fEventName.c_str(), tagname.c_str());
                     cm_msg_flush_buffer();
 
                     status = update_column(s->fEventName.c_str(), s->fTableName.c_str(), s->fColumnNames[j].c_str(), s->fColumnTypes[j].c_str(), s->fVariables[j].tag_name.c_str(), s->fVariables[i].tag_type.c_str(), timestamp, false, have_transaction);
                     if (status != HS_SUCCESS)
                        return status;
                  }
               }
            }
         }
 
         if (count == 0) {
            // tag does not have a corresponding column
            schema_ok = false;
            if (fDebug)
               printf("No column for tag %s!\n", tagname.c_str());
 
            bool found_column = false;
            
            if (write_enable) {
               for (size_t j=0; j<s->fVariables.size(); j++) {
                  if (tagname == s->fVariables[j].tag_name) {
                     bool typeok = s->fSql->TypesCompatible(tagtype, s->fColumnTypes[j].c_str());
                     if (typeok) {
                        cm_msg(MINFO, "SqlHistory::update_schema", "Reactivating SQL column \'%s\' type \'%s\' in table \"%s\" for history event \"%s\" tag \"%s\"", s->fColumnNames[j].c_str(), s->fColumnTypes[j].c_str(), s->fTableName.c_str(), s->fEventName.c_str(), tagname.c_str());
                        cm_msg_flush_buffer();
 
                        status = update_column(s->fEventName.c_str(), s->fTableName.c_str(), s->fColumnNames[j].c_str(), s->fColumnTypes[j].c_str(), s->fVariables[j].tag_name.c_str(), s->fVariables[j].tag_type.c_str(), timestamp, true, have_transaction);
                        if (status != HS_SUCCESS)
                           return status;
 
                        if (count == 0) {
                           s->fOffsets[j] = offset;
                           offset += rpc_tid_size(tagtype);
                        }
                        count++;
                        found_column = true;
                        if (count > 1) {
                           cm_msg(MERROR, "SqlHistory::update_schema", "Duplicate SQL column \'%s\' type \'%s\' in table \"%s\" for history event \"%s\" tag \"%s\"", s->fColumnNames[j].c_str(), s->fColumnTypes[j].c_str(), s->fTableName.c_str(), s->fEventName.c_str(), tagname.c_str());
                           cm_msg_flush_buffer();
                        }
                     }
                  }
               }
            }
 
            // create column
            if (!found_column && write_enable) {
               std::string col_name = maybe_colname;
               const char* col_type = s->fSql->ColumnType(tagtype);
 
               bool dupe = false;
               for (size_t kk=0; kk<s->fColumnNames.size(); kk++)
                  if (s->fColumnNames[kk] == col_name) {
                     dupe = true;
                     break;
                  }
 
               time_t now = time(NULL);
               
               bool retry = false;
               for (int t=0; t<20; t++) {
 
                  // if duplicate column name, change it, try again
                  if (dupe || retry) {
                     col_name = maybe_colname;
                     col_name += "_";
                     col_name += TimeToString(now);
                     if (t > 0) {
                        char s[256];
                        sprintf(s, "_%d", t);
                        col_name += s;
                     }
                  }
 
                  if (fDebug)
                     printf("SqlHistory::update_schema: table [%s], add column [%s] type [%s] for tag [%s]\n", s->fTableName.c_str(), col_name.c_str(), col_type, tagname.c_str());
 
                  status = CreateSqlColumn(fSql, s->fTableName.c_str(), col_name.c_str(), col_type, have_transaction, fDebug);
 
                  if (status == DB_KEY_EXIST) {
                     if (fDebug)
                        printf("SqlHistory::update_schema: table [%s], add column [%s] type [%s] for tag [%s] failed: duplicate column name\n", s->fTableName.c_str(), col_name.c_str(), col_type, tagname.c_str());
                     retry = true;
                     continue;
                  }
 
                  if (status != HS_SUCCESS)
                     return status;
 
                  break;
               }
 
               if (status != HS_SUCCESS)
                  return status;
 
               status = update_column(s->fEventName.c_str(), s->fTableName.c_str(), col_name.c_str(), col_type, tagname.c_str(), rpc_tid_name(tagtype), timestamp, true, have_transaction);
               if (status != HS_SUCCESS)
                  return status;
            }
         }
 
         if (count > 1) {
            // schema has duplicate tags
            schema_ok = false;
            cm_msg(MERROR, "SqlHistory::update_schema", "Duplicate tags or SQL columns for history event \"%s\" tag \"%s\"", s->fEventName.c_str(), tagname.c_str());
            cm_msg_flush_buffer();
         }
      }
   }
 
   // mark as unused all columns not listed in tags
 
   for (size_t k=0; k<s->fColumnNames.size(); k++)
      if (s->fVariables[k].name.length() > 0) {
         bool found = false;
 
         for (int i=0; i<ntags; i++) {
            for (unsigned int j=0; j<tags[i].n_data; j++) {
               std::string tagname = tags[i].name;
 
               if (tags[i].n_data > 1) {
                  char s[256];
                  sprintf(s, "[%d]", j);
                  tagname += s;
               }
 
               if (s->fVariables[k].name == tagname) {
                  found = true;
                  break;
               }
            }
 
            if (found)
               break;
         }
 
         if (!found) {
            // column not found in tags list
            schema_ok = false;
            if (fDebug)
               printf("Event [%s] Column [%s] tag [%s] not listed in tags list!\n", s->fEventName.c_str(), s->fColumnNames[k].c_str(), s->fVariables[k].name.c_str());
            if (write_enable) {
               cm_msg(MINFO, "SqlHistory::update_schema", "Deactivating SQL column \'%s\' type \'%s\' in table \"%s\" for history event \"%s\" not used for any tags", s->fColumnNames[k].c_str(), s->fColumnTypes[k].c_str(), s->fTableName.c_str(), s->fEventName.c_str());
               cm_msg_flush_buffer();
 
               status = update_column(s->fEventName.c_str(), s->fTableName.c_str(), s->fColumnNames[k].c_str(), s->fColumnTypes[k].c_str(), s->fVariables[k].tag_name.c_str(), s->fVariables[k].tag_type.c_str(), timestamp, false, have_transaction);
               if (status != HS_SUCCESS)
                  return status;
            }
         }
      }
 
   if (!write_enable)
      if (!schema_ok) {
         if (fDebug)
            printf("Return error!\n");
         return HS_FILE_ERROR;
      }
 
   return HS_SUCCESS;
}
 
//             SQLITE functions                       //
 
static int ReadSqliteTableNames(SqlBase* sql, HsSchemaVector *sv, const char* table_name, int debug)
{
   if (debug)
      printf("ReadSqliteTableNames: table [%s]\n", table_name);
 
   int status;
   std::string cmd;
 
   // FIXME: quotes
   cmd = "SELECT event_name, _i_time FROM \'_event_name_";
   cmd += table_name;
   cmd += "\' WHERE table_name='";
   cmd += table_name;
   cmd += "';";
 
   status = sql->Prepare(table_name, cmd.c_str());
 
   if (status != DB_SUCCESS)
      return status;
 
   while (1) {
      status = sql->Step();
 
      if (status != DB_SUCCESS)
         break;
 
      std::string xevent_name  = sql->GetText(0);
      time_t      xevent_time  = sql->GetTime(1);
 
      //printf("read event name [%s] time %s\n", xevent_name.c_str(), TimeToString(xevent_time).c_str());
 
      HsSqlSchema* s = new HsSqlSchema;
      s->fSql = sql;
      s->fEventName = xevent_name;
      s->fTimeFrom = xevent_time;
      s->fTimeTo = 0;
      s->fTableName = table_name;
      sv->add(s);
   }
 
   status = sql->Finalize();
 
   return HS_SUCCESS;
}
 
static int ReadSqliteTableSchema(SqlBase* sql, HsSchemaVector *sv, const char* table_name, int debug)
{
   if (debug)
      printf("ReadSqliteTableSchema: table [%s]\n", table_name);
 
   if (1) {
      // seed schema with table names
      HsSqlSchema* s = new HsSqlSchema;
      s->fSql = sql;
      s->fEventName = table_name;
      s->fTimeFrom = 0;
      s->fTimeTo = 0;
      s->fTableName = table_name;
      sv->add(s);
   }
 
   return ReadSqliteTableNames(sql, sv, table_name, debug);
}
 
//             SQLITE history classes                 //
 
class SqliteHistory: public SqlHistoryBase
{
public:
   SqliteHistory() { // ctor
#ifdef HAVE_SQLITE
      fSql = new Sqlite();
#endif
   }
 
   int read_table_and_event_names(HsSchemaVector *sv);
   int read_column_names(HsSchemaVector *sv, const char* table_name, const char* event_name);
   int create_table(HsSchemaVector* sv, const char* event_name, time_t timestamp);
   int update_column(const char* event_name, const char* table_name, const char* column_name, const char* column_type, const char* tag_name, const char* tag_type, const time_t timestamp, bool active, bool* have_transaction);
};
 
int SqliteHistory::read_table_and_event_names(HsSchemaVector *sv)
{
   int status;
 
   if (fDebug)
      printf("SqliteHistory::read_table_and_event_names!\n");
 
   // loop over all tables
 
   std::vector<std::string> tables;
   status = fSql->ListTables(&tables);
   if (status != DB_SUCCESS)
      return status;
 
   for (size_t i=0; i<tables.size(); i++) {
      const char* table_name = tables[i].c_str();
 
      const char* s;
      s = strstr(table_name, "_event_name_");
      if (s == table_name)
         continue;
      s = strstr(table_name, "_column_names_");
      if (s == table_name)
         continue;
 
      status = ReadSqliteTableSchema(fSql, sv, table_name, fDebug);
   }
 
   return HS_SUCCESS;
}
 
int SqliteHistory::read_column_names(HsSchemaVector *sv, const char* table_name, const char* event_name)
{
   if (fDebug)
      printf("SqliteHistory::read_column_names: table [%s], event [%s]\n", table_name, event_name);
 
   // for all schema for table_name, prepopulate is with column names
 
   std::vector<std::string> columns;
   fSql->ListColumns(table_name, &columns);
 
   // first, populate column names
 
   for (size_t i=0; i<sv->size(); i++) {
      HsSqlSchema* s = (HsSqlSchema*)(*sv)[i];
 
      if (s->fTableName != table_name)
         continue;
 
      // schema should be empty at this point
      //assert(s->fVariables.size() == 0);
 
      for (size_t j=0; j<columns.size(); j+=2) {
         const char* cn = columns[j+0].c_str();
         const char* ct = columns[j+1].c_str();
 
         if (strcmp(cn, "_t_time") == 0)
            continue;
         if (strcmp(cn, "_i_time") == 0)
            continue;
 
         bool found = (s->find_column_index(cn) >= 0);
 
         //printf("column [%s] sql type [%s]\n", cn.c_str(), ct);
 
         if (!found) {
            HsSchemaEntry se;
            se.name = cn;
            se.type = 0;
            se.n_data = 1;
            se.n_bytes = 0;
            s->fColumnIndexCache[cn] = s->fColumnNames.size();
            s->fVariables.push_back(se);
            s->fColumnNames.push_back(cn);
            s->fColumnTypes.push_back(ct);
            s->fColumnInactive.push_back(false);
            s->fOffsets.push_back(-1);
         }
      }
   }
 
   // then read column name information
 
   std::string tn;
   tn += "_column_names_";
   tn += table_name;
   
   std::string cmd;
   cmd = "SELECT column_name, tag_name, tag_type, _i_time FROM ";
   cmd += fSql->QuoteId(tn.c_str());
   cmd += " WHERE table_name=";
   cmd += fSql->QuoteString(table_name);
   cmd += " ORDER BY _i_time ASC;";
 
   int status = fSql->Prepare(table_name, cmd.c_str());
 
   if (status != DB_SUCCESS) {
      return status;
   }
 
   while (1) {
      status = fSql->Step();
 
      if (status != DB_SUCCESS)
         break;
 
      // NOTE: SQL "SELECT ORDER BY _i_time ASC" returns data sorted by time
      // in this code we use the data from the last data row
      // so if multiple rows are present, the latest one is used
 
      std::string col_name  = fSql->GetText(0);
      std::string tag_name  = fSql->GetText(1);
      std::string tag_type  = fSql->GetText(2);
      time_t   schema_time  = fSql->GetTime(3);
 
      //printf("read table [%s] column [%s] tag name [%s] time %s\n", table_name, col_name.c_str(), tag_name.c_str(), TimeToString(xxx_time).c_str());
 
      // make sure a schema exists at this time point
      NewSqlSchema(sv, table_name, schema_time);
 
      // add this information to all schema
 
      for (size_t i=0; i<sv->size(); i++) {
         HsSqlSchema* s = (HsSqlSchema*)(*sv)[i];
         if (s->fTableName != table_name)
            continue;
         if (s->fTimeFrom < schema_time)
            continue;
 
         //printf("add column to schema %d\n", s->fTimeFrom);
 
         int j = s->find_column_index(col_name);
         if (j < 0)
            continue;
         s->fVariables[j].name = tag_name;
         s->fVariables[j].type = rpc_name_tid(tag_type.c_str());
         s->fVariables[j].n_data = 1;
         s->fVariables[j].n_bytes = rpc_tid_size(s->fVariables[j].type);
      }
   }
 
   status = fSql->Finalize();
 
   return HS_SUCCESS;
}
 
int SqliteHistory::create_table(HsSchemaVector* sv, const char* event_name, time_t timestamp)
{
   if (fDebug)
      printf("SqliteHistory::create_table: event [%s], timestamp %s\n", event_name, TimeToString(timestamp).c_str());
 
   int status;
   bool have_transaction = false;
   std::string table_name = MidasNameToSqlName(event_name);
 
   // FIXME: what about duplicate table names?
   status = CreateSqlTable(fSql, table_name.c_str(), &have_transaction);
 
   //if (status == DB_KEY_EXIST) {
   //   return ReadSqliteTableSchema(fSql, sv, table_name.c_str(), fDebug);
   //}
 
   if (status != HS_SUCCESS) {
      // FIXME: ???
      // FIXME: at least close or revert the transaction
      return status;
   }
 
   std::string cmd;
 
   std::string en;
   en += "_event_name_";
   en += table_name;
 
   cmd = "CREATE TABLE ";
   cmd += fSql->QuoteId(en.c_str());
   cmd += " (table_name TEXT NOT NULL, event_name TEXT NOT NULL, _i_time INTEGER NOT NULL);";
 
   status = fSql->Exec(table_name.c_str(), cmd.c_str());
 
   cmd = "INSERT INTO ";
   cmd += fSql->QuoteId(en.c_str());
   cmd += " (table_name, event_name, _i_time) VALUES (";
   cmd += fSql->QuoteString(table_name.c_str());
   cmd += ", ";
   cmd += fSql->QuoteString(event_name);
   cmd += ", ";
   cmd += fSql->QuoteString(TimeToString(timestamp).c_str());
   cmd += ");";
 
   status = fSql->Exec(table_name.c_str(), cmd.c_str());
 
   std::string cn;
   cn += "_column_names_";
   cn += table_name;
 
   cmd = "CREATE TABLE ";
   cmd += fSql->QuoteId(cn.c_str());
   cmd += " (table_name TEXT NOT NULL, column_name TEXT NOT NULL, tag_name TEXT NOT NULL, tag_type TEXT NOT NULL, column_type TEXT NOT NULL, _i_time INTEGER NOT NULL);";
 
   status = fSql->Exec(table_name.c_str(), cmd.c_str());
 
   status = fSql->CommitTransaction(table_name.c_str());
   if (status != DB_SUCCESS) {
      return HS_FILE_ERROR;
   }
 
   return ReadSqliteTableSchema(fSql, sv, table_name.c_str(), fDebug);
}
 
int SqliteHistory::update_column(const char* event_name, const char* table_name, const char* column_name, const char* column_type, const char* tag_name, const char* tag_type, const time_t timestamp, bool active, bool* have_transaction)
{
   if (fDebug)
      printf("SqliteHistory::update_column: event [%s], table [%s], column [%s], new name [%s], timestamp %s\n", event_name, table_name, column_name, tag_name, TimeToString(timestamp).c_str());
 
   int status = StartSqlTransaction(fSql, table_name, have_transaction);
   if (status != HS_SUCCESS)
      return status;
 
   // FIXME: quotes
   std::string cmd;
   cmd = "INSERT INTO \'_column_names_";
   cmd += table_name;
   cmd += "\' (table_name, column_name, tag_name, tag_type, column_type, _i_time) VALUES (\'";
   cmd += table_name;
   cmd += "\', \'";
   cmd += column_name;
   cmd += "\', \'";
   cmd += tag_name;
   cmd += "\', \'";
   cmd += tag_type;
   cmd += "\', \'";
   cmd += column_type;
   cmd += "\', \'";
   cmd += TimeToString(timestamp);
   cmd += "\');";
   status = fSql->Exec(table_name, cmd.c_str());
 
   return status;
}
 
//              Mysql history classes                 //
 
class MysqlHistory: public SqlHistoryBase
{
public:
   MysqlHistory() { // ctor
#ifdef HAVE_MYSQL
      fSql = new Mysql();
#endif
   }
 
   int read_table_and_event_names(HsSchemaVector *sv);
   int read_column_names(HsSchemaVector *sv, const char* table_name, const char* event_name);
   int create_table(HsSchemaVector* sv, const char* event_name, time_t timestamp);
   int update_column(const char* event_name, const char* table_name, const char* column_name, const char* column_type, const char* tag_name, const char* tag_type, const time_t timestamp, bool active, bool* have_transaction);
};
 
static int ReadMysqlTableNames(SqlBase* sql, HsSchemaVector *sv, const char* table_name, int debug, const char* must_have_event_name, const char* must_have_table_name)
{
   if (debug)
      printf("ReadMysqlTableNames: table [%s], must have event [%s] table [%s]\n", table_name, must_have_event_name, must_have_table_name);
 
   int status;
   std::string cmd;
 
   if (table_name) {
      cmd = "SELECT event_name, table_name, itimestamp FROM _history_index WHERE table_name='";
      cmd += table_name;
      cmd += "';";
   } else {
      // Only enumerate the table-marker rows (one per table/version), not the far more
      // numerous per-column rows. Both carry a non-empty table_name, but the table-marker
      // INSERT omits column_name (leaving it NULL) while the column-record INSERT sets it.
      // Restricting to column_name IS NULL keeps the reader schema vector at ~(#tables)
      // instead of ~(#columns x #versions), which is what blows up on a large _history_index.
      cmd = "SELECT event_name, table_name, itimestamp FROM _history_index WHERE table_name!='' AND column_name IS NULL;";
      table_name = "_history_index";
   }
 
   status = sql->Prepare(table_name, cmd.c_str());
 
   if (status != DB_SUCCESS)
      return status;
 
   bool found_must_have_table = false;
   int count = 0;
 
   while (1) {
      status = sql->Step();
 
      if (status != DB_SUCCESS)
         break;
 
      const char* xevent_name  = sql->GetText(0);
      const char* xtable_name  = sql->GetText(1);
      time_t      xevent_time  = sql->GetTime(2);
 
      if (debug == 999) {
         printf("entry %d event name [%s] table name [%s] time %s\n", count, xevent_name, xtable_name, TimeToString(xevent_time).c_str());
      }
 
      if (must_have_table_name && (strcmp(xtable_name, must_have_table_name) == 0)) {
         assert(must_have_event_name != NULL);
         if (event_name_cmp(xevent_name, must_have_event_name) == 0) {
            found_must_have_table = true;
            //printf("Found table [%s]: event name [%s] table name [%s] time %s\n", must_have_table_name, xevent_name, xtable_name, TimeToString(xevent_time).c_str());
         } else {
            //printf("Found correct table [%s] with wrong event name [%s] expected [%s] time %s\n", must_have_table_name, xevent_name, must_have_event_name, TimeToString(xevent_time).c_str());
         }
      }
      
      HsSqlSchema* s = new HsSqlSchema;
      s->fSql = sql;
      s->fEventName = xevent_name;
      s->fTimeFrom = xevent_time;
      s->fTimeTo = 0;
      s->fTableName = xtable_name;
      sv->add(s);
      count++;
   }
 
   status = sql->Finalize();
 
   if (must_have_table_name && !found_must_have_table) {
      cm_msg(MERROR, "ReadMysqlTableNames", "Error: Table [%s] for event [%s] missing from the history index\n", must_have_table_name, must_have_event_name);
      if (debug == 999)
         return HS_FILE_ERROR;
      // NB: recursion is broken by setting debug to 999.
      ReadMysqlTableNames(sql, sv, table_name, 999, must_have_event_name, must_have_table_name);
      cm_msg(MERROR, "ReadMysqlTableNames", "Error: Cannot continue, nothing will work after this error\n");
      cm_msg_flush_buffer();
      abort();
      return HS_FILE_ERROR;
   }
 
   if (0) {
      // print accumulated schema
      printf("ReadMysqlTableNames: table_name [%s] event_name [%s] table_name [%s]\n", table_name, must_have_event_name, must_have_table_name);
      sv->print(false);
   }
 
   return HS_SUCCESS;
}
 
int MysqlHistory::read_column_names(HsSchemaVector *sv, const char* table_name, const char* event_name)
{
   if (fDebug)
      printf("MysqlHistory::read_column_names: table [%s], event [%s]\n", table_name, event_name);
 
   // for all schema for table_name, prepopulate is with column names
 
   std::vector<std::string> columns;
   fSql->ListColumns(table_name, &columns);
 
   // first, populate column names
 
   for (size_t i=0; i<sv->size(); i++) {
      HsSqlSchema* s = (HsSqlSchema*)(*sv)[i];
 
      if (s->fTableName != table_name)
         continue;
 
      // schema should be empty at this point
      //assert(s->fVariables.size() == 0);
 
      for (size_t j=0; j<columns.size(); j+=2) {
         const char* cn = columns[j+0].c_str();
         const char* ct = columns[j+1].c_str();
 
         if (strcmp(cn, "_t_time") == 0)
            continue;
         if (strcmp(cn, "_i_time") == 0)
            continue;
 
         bool found = (s->find_column_index(cn) >= 0);
 
         //printf("column [%s] sql type [%s]\n", cn.c_str(), ct);
 
         if (!found) {
            HsSchemaEntry se;
            se.tag_name = cn;
            se.tag_type = "";
            se.name = cn;
            se.type = 0;
            se.n_data = 1;
            se.n_bytes = 0;
            s->fColumnIndexCache[cn] = s->fColumnNames.size();
            s->fVariables.push_back(se);
            s->fColumnNames.push_back(cn);
            s->fColumnTypes.push_back(ct);
            s->fColumnInactive.push_back(false);
            s->fOffsets.push_back(-1);
         }
      }
   }
 
   // then read column name information
 
   std::string cmd;
   cmd = "SELECT column_name, column_type, tag_name, tag_type, itimestamp, active FROM _history_index WHERE event_name=";
   cmd += fSql->QuoteString(event_name);
   cmd += ";";
 
   int status = fSql->Prepare(table_name, cmd.c_str());
 
   if (status != DB_SUCCESS) {
      return status;
   }
 
   while (1) {
      status = fSql->Step();
 
      if (status != DB_SUCCESS)
         break;
 
      const char* col_name  = fSql->GetText(0);
      const char* col_type  = fSql->GetText(1);
      const char* tag_name  = fSql->GetText(2);
      const char* tag_type  = fSql->GetText(3);
      time_t   schema_time  = fSql->GetTime(4);
      const char* active    = fSql->GetText(5);
      int iactive = atoi(active);
 
      //printf("read table [%s] column [%s] type [%s] tag name [%s] type [%s] time %s active [%s] %d\n", table_name, col_name, col_type, tag_name, tag_type, TimeToString(schema_time).c_str(), active, iactive);
 
      if (!col_name)
         continue;
      if (!tag_name)
         continue;
      if (strlen(col_name) < 1)
         continue;
 
      // make sure a schema exists at this time point
      NewSqlSchema(sv, table_name, schema_time);
 
      // add this information to all schema
 
      for (size_t i=0; i<sv->size(); i++) {
         HsSqlSchema* s = (HsSqlSchema*)(*sv)[i];
         if (s->fTableName != table_name)
            continue;
         if (s->fTimeFrom < schema_time)
            continue;
 
         int tid = rpc_name_tid(tag_type);
         int tid_size = rpc_tid_size(tid);
 
         int j = s->find_column_index(col_name);
         if (j < 0)
            continue;
 
         s->fVariables[j].tag_name = tag_name;
         s->fVariables[j].tag_type = tag_type;
         if (!iactive) {
            s->fVariables[j].name = "";
            s->fColumnInactive[j] = true;
         } else {
            s->fVariables[j].name = tag_name;
            s->fColumnInactive[j] = false;
         }
         s->fVariables[j].type = tid;
         s->fVariables[j].n_data = 1;
         s->fVariables[j].n_bytes = tid_size;
 
         // doctor column names in case MySQL returns different type
         // from the type used to create the column, but the types
         // are actually the same. K.O.
         DoctorSqlColumnType(&s->fColumnTypes[j], col_type);
      }
   }
 
   status = fSql->Finalize();
 
   return HS_SUCCESS;
}
 
#if 0
static int ReadMysqlTableSchema(SqlBase* sql, HsSchemaVector *sv, const char* table_name, int debug)
{
   if (debug)
      printf("ReadMysqlTableSchema: table [%s]\n", table_name);
 
   if (1) {
      // seed schema with table names
      HsSqlSchema* s = new HsSqlSchema;
      s->fSql = sql;
      s->fEventName = table_name;
      s->fTimeFrom = 0;
      s->fTimeTo = 0;
      s->fTableName = table_name;
      sv->add(s);
   }
 
   return ReadMysqlTableNames(sql, sv, table_name, debug, NULL, NULL);
}
#endif
 
int MysqlHistory::read_table_and_event_names(HsSchemaVector *sv)
{
   int status;
 
   if (fDebug)
      printf("MysqlHistory::read_table_and_event_names!\n");
 
   // loop over all tables
 
   std::vector<std::string> tables;
   status = fSql->ListTables(&tables);
   if (status != DB_SUCCESS)
      return status;
 
   for (size_t i=0; i<tables.size(); i++) {
      const char* table_name = tables[i].c_str();
 
      const char* s;
      s = strstr(table_name, "_history_index");
      if (s == table_name)
         continue;
 
      if (1) {
         // seed schema with table names
         HsSqlSchema* s = new HsSqlSchema;
         s->fSql = fSql;
         s->fEventName = table_name;
         s->fTimeFrom = 0;
         s->fTimeTo = 0;
         s->fTableName = table_name;
         sv->add(s);
      }
   }
 
   if (0) {
      // print accumulated schema
      printf("read_table_and_event_names:\n");
      sv->print(false);
   }
 
   status = ReadMysqlTableNames(fSql, sv, NULL, fDebug, NULL, NULL);
 
   return HS_SUCCESS;
}
 
int MysqlHistory::create_table(HsSchemaVector* sv, const char* event_name, time_t timestamp)
{
   if (fDebug)
      printf("MysqlHistory::create_table: event [%s], timestamp %s\n", event_name, TimeToString(timestamp).c_str());
 
   int status;
   std::string table_name = MidasNameToSqlName(event_name);
 
   // MySQL table name length limit is 64 bytes
   if (table_name.length() > 40) {
      table_name.resize(40);
      table_name += "_T";
   }
 
   time_t now = time(NULL);
 
   int max_attempts = 10;
   for (int i=0; i<max_attempts; i++) {
      status = fSql->OpenTransaction(table_name.c_str());
      if (status != DB_SUCCESS) {
         return HS_FILE_ERROR;
      }
 
      bool have_transaction = true;
 
      std::string xtable_name = table_name;
 
      if (i>0) {
         xtable_name += "_";
         xtable_name += TimeToString(now);
         if (i>1) {
            xtable_name += "_";
            char buf[256];
            sprintf(buf, "%d", i);
            xtable_name += buf;
         }
      }
      
      status = CreateSqlTable(fSql, xtable_name.c_str(), &have_transaction);
 
      //printf("event [%s] create table [%s] status %d\n", event_name, xtable_name.c_str(), status);
 
      if (status == DB_KEY_EXIST) {
         // already exists, try with different name!
         fSql->RollbackTransaction(table_name.c_str());
         continue;
      }
 
      if (status != HS_SUCCESS) {
         // MYSQL cannot roll back "create table", if we cannot create SQL tables, nothing will work. Give up now.
         cm_msg(MERROR, "MysqlHistory::create_table", "Could not create table [%s] for event [%s], timestamp %s, please fix the SQL database configuration and try again", table_name.c_str(), event_name, TimeToString(timestamp).c_str());
         abort();
 
         // fatal error, give up!
         fSql->RollbackTransaction(table_name.c_str());
         break;
      }
 
      for (int j=0; j<2; j++) {
         std::string cmd;
         cmd += "INSERT INTO _history_index (event_name, table_name, itimestamp, active) VALUES (";
         cmd += fSql->QuoteString(event_name);
         cmd += ", ";
         cmd += fSql->QuoteString(xtable_name.c_str());
         cmd += ", ";
         char buf[256];
         sprintf(buf, "%.0f", (double)timestamp);
         cmd += fSql->QuoteString(buf);
         cmd += ", ";
         cmd += fSql->QuoteString("1");
         cmd += ");";
         
         int status = fSql->Exec(table_name.c_str(), cmd.c_str());
         if (status == DB_SUCCESS)
            break;
         
         status = CreateSqlTable(fSql,  "_history_index", &have_transaction);
         status = CreateSqlColumn(fSql, "_history_index", "event_name",  "varchar(256) character set binary not null", &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "table_name",  "varchar(256)",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "tag_name",    "varchar(256) character set binary",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "tag_type",    "varchar(256)",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "column_name", "varchar(256)",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "column_type", "varchar(256)",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "itimestamp",  "integer not null", &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "active",      "boolean",          &have_transaction, fDebug);
      }
 
      status = fSql->CommitTransaction(table_name.c_str());
 
      if (status != DB_SUCCESS) {
         return HS_FILE_ERROR;
      }
      
      return ReadMysqlTableNames(fSql, sv, xtable_name.c_str(), fDebug, event_name, xtable_name.c_str());
   }
 
   cm_msg(MERROR, "MysqlHistory::create_table", "Could not create table [%s] for event [%s], timestamp %s, after %d attempts", table_name.c_str(), event_name, TimeToString(timestamp).c_str(), max_attempts);
 
   return HS_FILE_ERROR;
}
 
int MysqlHistory::update_column(const char* event_name, const char* table_name, const char* column_name, const char* column_type, const char* tag_name, const char* tag_type, const time_t timestamp, bool active, bool* have_transaction)
{
   if (fDebug)
      printf("MysqlHistory::update_column: event [%s], table [%s], column [%s], type [%s] new name [%s], timestamp %s\n", event_name, table_name, column_name, column_type, tag_name, TimeToString(timestamp).c_str());
 
   std::string cmd;
   cmd += "INSERT INTO _history_index (event_name, table_name, tag_name, tag_type, column_name, column_type, itimestamp, active) VALUES (";
   cmd += fSql->QuoteString(event_name);
   cmd += ", ";
   cmd += fSql->QuoteString(table_name);
   cmd += ", ";
   cmd += fSql->QuoteString(tag_name);
   cmd += ", ";
   cmd += fSql->QuoteString(tag_type);
   cmd += ", ";
   cmd += fSql->QuoteString(column_name);
   cmd += ", ";
   cmd += fSql->QuoteString(column_type);
   cmd += ", ";
   char buf[256];
   sprintf(buf, "%.0f", (double)timestamp);
   cmd += fSql->QuoteString(buf);
   cmd += ", ";
   if (active)
      cmd += fSql->QuoteString("1");
   else
      cmd += fSql->QuoteString("0");
   cmd += ");";
         
   int status = fSql->Exec(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   return HS_SUCCESS;
}
 
//         PostgreSQL history classes                 //
 
#ifdef HAVE_PGSQL
 
class PgsqlHistory: public SqlHistoryBase
{
public:
   Pgsql *fPgsql = NULL;
public:
   PgsqlHistory() { // ctor
      fPgsql = new Pgsql();
      fSql = fPgsql;
   }
 
   int read_table_and_event_names(HsSchemaVector *sv);
   int read_column_names(HsSchemaVector *sv, const char* table_name, const char* event_name);
   int create_table(HsSchemaVector* sv, const char* event_name, time_t timestamp);
   int update_column(const char* event_name, const char* table_name, const char* column_name, const char* column_type, const char* tag_name, const char* tag_type, const time_t timestamp, bool active, bool* have_transaction);
};
 
static int ReadPgsqlTableNames(SqlBase* sql, HsSchemaVector *sv, const char* table_name, int debug, const char* must_have_event_name, const char* must_have_table_name)
{
   if (debug)
      printf("ReadPgsqlTableNames: table [%s], must have event [%s] table [%s]\n", table_name, must_have_event_name, must_have_table_name);
 
   int status;
   std::string cmd;
 
   if (table_name) {
      cmd = "SELECT event_name, table_name, itimestamp FROM _history_index WHERE table_name='";
      cmd += table_name;
      cmd += "';";
   } else {
      // Only enumerate the table-marker rows (one per table/version), not the far more
      // numerous per-column rows. Both carry a non-empty table_name, but the table-marker
      // INSERT omits column_name (leaving it NULL) while the column-record INSERT sets it.
      // Restricting to column_name IS NULL keeps the reader schema vector at ~(#tables)
      // instead of ~(#columns x #versions), which is what blows up on a large _history_index.
      cmd = "SELECT event_name, table_name, itimestamp FROM _history_index WHERE table_name!='' AND column_name IS NULL;";
      table_name = "_history_index";
   }
 
   status = sql->Prepare(table_name, cmd.c_str());
 
   if (status != DB_SUCCESS)
      return status;
 
   bool found_must_have_table = false;
   int count = 0;
 
   while (1) {
      status = sql->Step();
 
      if (status != DB_SUCCESS)
         break;
 
      const char* xevent_name  = sql->GetText(0);
      const char* xtable_name  = sql->GetText(1);
      time_t      xevent_time  = sql->GetTime(2);
 
      if (debug == 999) {
         printf("entry %d event name [%s] table name [%s] time %s\n", count, xevent_name, xtable_name, TimeToString(xevent_time).c_str());
      }
 
      if (must_have_table_name && (strcmp(xtable_name, must_have_table_name) == 0)) {
         assert(must_have_event_name != NULL);
         if (event_name_cmp(xevent_name, must_have_event_name) == 0) {
            found_must_have_table = true;
            //printf("Found table [%s]: event name [%s] table name [%s] time %s\n", must_have_table_name, xevent_name, xtable_name, TimeToString(xevent_time).c_str());
         } else {
            //printf("Found correct table [%s] with wrong event name [%s] expected [%s] time %s\n", must_have_table_name, xevent_name, must_have_event_name, TimeToString(xevent_time).c_str());
         }
      }
      
      HsSqlSchema* s = new HsSqlSchema;
      s->fSql = sql;
      s->fEventName = xevent_name;
      s->fTimeFrom = xevent_time;
      s->fTimeTo = 0;
      s->fTableName = xtable_name;
      sv->add(s);
      count++;
   }
 
   status = sql->Finalize();
 
   if (must_have_table_name && !found_must_have_table) {
      cm_msg(MERROR, "ReadPgsqlTableNames", "Error: Table [%s] for event [%s] missing from the history index\n", must_have_table_name, must_have_event_name);
      if (debug == 999)
         return HS_FILE_ERROR;
      // NB: recursion is broken by setting debug to 999.
      ReadPgsqlTableNames(sql, sv, table_name, 999, must_have_event_name, must_have_table_name);
      cm_msg(MERROR, "ReadPgsqlTableNames", "Error: Cannot continue, nothing will work after this error\n");
      cm_msg_flush_buffer();
      abort();
      return HS_FILE_ERROR;
   }
 
   if (0) {
      // print accumulated schema
      printf("ReadPgsqlTableNames: table_name [%s] event_name [%s] table_name [%s]\n", table_name, must_have_event_name, must_have_table_name);
      sv->print(false);
   }
 
   return HS_SUCCESS;
}
 
int PgsqlHistory::read_column_names(HsSchemaVector *sv, const char* table_name, const char* event_name)
{
   if (fDebug)
      printf("PgsqlHistory::read_column_names: table [%s], event [%s]\n", table_name, event_name);
 
   // for all schema for table_name, prepopulate is with column names
 
   std::vector<std::string> columns;
   fSql->ListColumns(table_name, &columns);
 
   // first, populate column names
 
   for (size_t i=0; i<sv->size(); i++) {
      HsSqlSchema* s = (HsSqlSchema*)(*sv)[i];
 
      if (s->fTableName != table_name)
         continue;
 
      // schema should be empty at this point
      //assert(s->fVariables.size() == 0);
 
      for (size_t j=0; j<columns.size(); j+=2) {
         const char* cn = columns[j+0].c_str();
         const char* ct = columns[j+1].c_str();
 
         if (strcmp(cn, "_t_time") == 0)
            continue;
         if (strcmp(cn, "_i_time") == 0)
            continue;
 
         bool found = (s->find_column_index(cn) >= 0);
 
         if (!found) {
            HsSchemaEntry se;
            se.tag_name = cn;
            se.tag_type = "";
            se.name = cn;
            se.type = 0;
            se.n_data = 1;
            se.n_bytes = 0;
            s->fColumnIndexCache[cn] = s->fColumnNames.size();
            s->fVariables.push_back(se);
            s->fColumnNames.push_back(cn);
            s->fColumnTypes.push_back(ct);
            s->fColumnInactive.push_back(false);
            s->fOffsets.push_back(-1);
         }
      }
   }
 
   // then read column name information
 
   std::string cmd;
   cmd = "SELECT column_name, column_type, tag_name, tag_type, itimestamp, active FROM _history_index WHERE event_name=";
   cmd += fSql->QuoteString(event_name);
   cmd += ";";
 
   int status = fSql->Prepare(table_name, cmd.c_str());
 
   if (status != DB_SUCCESS) {
      return status;
   }
 
   while (1) {
      status = fSql->Step();
 
      if (status != DB_SUCCESS)
         break;
 
      const char* col_name  = fSql->GetText(0);
      const char* col_type  = fSql->GetText(1);
      const char* tag_name  = fSql->GetText(2);
      const char* tag_type  = fSql->GetText(3);
      time_t   schema_time  = fSql->GetTime(4);
      const char* active    = fSql->GetText(5);
      int iactive = atoi(active);
 
      //printf("read table [%s] column [%s] type [%s] tag name [%s] type [%s] time %s active [%s] %d\n", table_name, col_name, col_type, tag_name, tag_type, TimeToString(schema_time).c_str(), active, iactive);
 
      if (!col_name)
         continue;
      if (!tag_name)
         continue;
      if (strlen(col_name) < 1)
         continue;
 
      // make sure a schema exists at this time point
      NewSqlSchema(sv, table_name, schema_time);
 
      // add this information to all schema
      for (size_t i=0; i<sv->size(); i++) {
         HsSqlSchema* s = (HsSqlSchema*)(*sv)[i];
         if (s->fTableName != table_name)
            continue;
         if (s->fTimeFrom < schema_time)
            continue;
 
         int tid = rpc_name_tid(tag_type);
         int tid_size = rpc_tid_size(tid);
 
         int j = s->find_column_index(col_name);
         if (j < 0)
            continue;
 
         s->fVariables[j].tag_name = tag_name;
         s->fVariables[j].tag_type = tag_type;
         if (!iactive) {
            s->fVariables[j].name = "";
            s->fColumnInactive[j] = true;
         } else {
            s->fVariables[j].name = tag_name;
            s->fColumnInactive[j] = false;
         }
         s->fVariables[j].type = tid;
         s->fVariables[j].n_data = 1;
         s->fVariables[j].n_bytes = tid_size;
 
         // doctor column names in case MySQL returns different type
         // from the type used to create the column, but the types
         // are actually the same. K.O.
         DoctorPgsqlColumnType(&s->fColumnTypes[j], col_type);
      }
   }
 
   status = fSql->Finalize();
 
   return HS_SUCCESS;
}
 
int PgsqlHistory::read_table_and_event_names(HsSchemaVector *sv)
{
   int status;
 
   if (fDebug)
      printf("PgsqlHistory::read_table_and_event_names!\n");
 
   // loop over all tables
 
   std::vector<std::string> tables;
   status = fSql->ListTables(&tables);
   if (status != DB_SUCCESS)
      return status;
 
   for (size_t i=0; i<tables.size(); i++) {
      const char* table_name = tables[i].c_str();
 
      const char* s;
      s = strstr(table_name, "_history_index");
      if (s == table_name)
         continue;
 
      if (1) {
         // seed schema with table names
         HsSqlSchema* s = new HsSqlSchema;
         s->fSql = fSql;
         s->fEventName = table_name;
         s->fTimeFrom = 0;
         s->fTimeTo = 0;
         s->fTableName = table_name;
         sv->add(s);
      }
   }
 
   if (0) {
      // print accumulated schema
      printf("read_table_and_event_names:\n");
      sv->print(false);
   }
 
   status = ReadPgsqlTableNames(fSql, sv, NULL, fDebug, NULL, NULL);
 
   return HS_SUCCESS;
}
 
int PgsqlHistory::create_table(HsSchemaVector* sv, const char* event_name, time_t timestamp)
{
   if (fDebug)
      printf("PgsqlHistory::create_table: event [%s], timestamp %s\n", event_name, TimeToString(timestamp).c_str());
 
   int status;
   std::string table_name = MidasNameToSqlName(event_name);
 
   // PostgreSQL table name length limit is 64 bytes
   if (table_name.length() > 40) {
      table_name.resize(40);
      table_name += "_T";
   }
 
   time_t now = time(NULL);
 
   int max_attempts = 10;
   for (int i=0; i<max_attempts; i++) {
      status = fSql->OpenTransaction(table_name.c_str());
      if (status != DB_SUCCESS) {
         return HS_FILE_ERROR;
      }
 
      bool have_transaction = true;
 
      std::string xtable_name = table_name;
 
      if (i>0) {
         xtable_name += "_";
         xtable_name += TimeToString(now);
         if (i>1) {
            xtable_name += "_";
            char buf[256];
            sprintf(buf, "%d", i);
            xtable_name += buf;
         }
      }
      
      if (fPgsql->fDownsample)
         status = CreateSqlHyperTable(fSql, xtable_name.c_str(), &have_transaction);
      else 
         status = CreateSqlTable(fSql, xtable_name.c_str(), &have_transaction);
 
      //printf("event [%s] create table [%s] status %d\n", event_name, xtable_name.c_str(), status);
 
      if (status == DB_KEY_EXIST) {
         // already exists, try with different name!
         fSql->RollbackTransaction(table_name.c_str());
         continue;
      }
 
      if (status != HS_SUCCESS) {
         fSql->RollbackTransaction(table_name.c_str());
         continue;
      }
 
      fSql->Exec(table_name.c_str(), "SAVEPOINT t0");
 
      for (int j=0; j<2; j++) {
         std::string cmd;
         cmd += "INSERT INTO _history_index (event_name, table_name, itimestamp, active) VALUES (";
         cmd += fSql->QuoteString(event_name);
         cmd += ", ";
         cmd += fSql->QuoteString(xtable_name.c_str());
         cmd += ", ";
         char buf[256];
         sprintf(buf, "%.0f", (double)timestamp);
         cmd += buf;
         cmd += ", ";
         cmd += fSql->QuoteString("1");
         cmd += ");";
         
         int status = fSql->Exec(table_name.c_str(), cmd.c_str());
         if (status == DB_SUCCESS)
            break;
         
         // if INSERT failed _history_index does not exist then recover to savepoint t0
         // to prevent whole transition abort
         fSql->Exec(table_name.c_str(), "ROLLBACK TO SAVEPOINT t0");
 
         status = CreateSqlTable(fSql,  "_history_index", &have_transaction, true);
         status = CreateSqlColumn(fSql, "_history_index", "event_name",  "text not null", &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "table_name",  "text",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "tag_name",    "text",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "tag_type",    "text",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "column_name", "text",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "column_type", "text",     &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "itimestamp",  "integer not null", &have_transaction, fDebug);
         status = CreateSqlColumn(fSql, "_history_index", "active",      "smallint", &have_transaction, fDebug);
 
         status = fSql->CommitTransaction(table_name.c_str());
      }
 
      if (status != DB_SUCCESS) {
         return HS_FILE_ERROR;
      }
      
      return ReadPgsqlTableNames(fSql, sv, xtable_name.c_str(), fDebug, event_name, xtable_name.c_str());
   }
 
   cm_msg(MERROR, "PgsqlHistory::create_table", "Could not create table [%s] for event [%s], timestamp %s, after %d attempts", table_name.c_str(), event_name, TimeToString(timestamp).c_str(), max_attempts);
 
   return HS_FILE_ERROR;
}
 
int PgsqlHistory::update_column(const char* event_name, const char* table_name, const char* column_name, const char* column_type, const char* tag_name, const char* tag_type, const time_t timestamp, bool active, bool* have_transaction)
{
   if (fDebug)
      printf("PgsqlHistory::update_column: event [%s], table [%s], column [%s], type [%s] new name [%s], timestamp %s\n", event_name, table_name, column_name, column_type, tag_name, TimeToString(timestamp).c_str());
 
   std::string cmd;
   cmd += "INSERT INTO _history_index (event_name, table_name, tag_name, tag_type, column_name, column_type, itimestamp, active) VALUES (";
   cmd += fSql->QuoteString(event_name);
   cmd += ", ";
   cmd += fSql->QuoteString(table_name);
   cmd += ", ";
   cmd += fSql->QuoteString(tag_name);
   cmd += ", ";
   cmd += fSql->QuoteString(tag_type);
   cmd += ", ";
   cmd += fSql->QuoteString(column_name);
   cmd += ", ";
   cmd += fSql->QuoteString(column_type);
   cmd += ", ";
   char buf[256];
   sprintf(buf, "%.0f", (double)timestamp);
   cmd += buf;
   cmd += ", ";
   if (active)
      cmd += fSql->QuoteString("1");
   else
      cmd += fSql->QuoteString("0");
   cmd += ");";
         
   int status = fSql->Exec(table_name, cmd.c_str());
   if (status != DB_SUCCESS)
      return HS_FILE_ERROR;
 
   return HS_SUCCESS;
}
 
#endif // HAVE_PGSQL
 
//               File history class                   //
 
const time_t kDay = 24*60*60;
const time_t kMonth = 30*kDay;
 
const double KiB = 1024;
const double MiB = KiB*KiB;
//const double GiB = KiB*MiB;
 
class FileHistory: public SchemaHistoryBase
{
protected:
   std::string fPath;
   time_t fPathLastMtime = 0;
   std::vector<std::string> fSortedFiles;
   std::vector<bool>        fSortedRead;
   time_t  fConfMaxFileAge  = 1*kMonth;
   off64_t fConfMaxFileSize = 100*MiB;
 
public:
   FileHistory() // ctor
   {
      // empty
   }
 
   int hs_connect(const char* connect_string);
   int hs_disconnect();
   int hs_clear_cache();
   int read_schema(HsSchemaVector* sv, const char* event_name, const time_t timestamp);
   HsSchema* new_event(const char* event_name, time_t timestamp, int ntags, const TAG tags[]);
 
protected:
   int create_file(const char* event_name, time_t timestamp, const std::vector<HsSchemaEntry>& vars, std::string* filenamep);
   HsFileSchema* read_file_schema(const char* filename);
   int read_file_list(bool *pchanged);
   void clear_file_list();
   void tags_to_variables(int ntags, const TAG tags[], std::vector<HsSchemaEntry>& variables);
   HsSchema* maybe_reopen(const char* event_name, time_t timestamp, HsSchema* s);
};
 
int FileHistory::hs_connect(const char* connect_string)
{
   if (fDebug)
      printf("hs_connect [%s]!\n", connect_string);
 
   hs_disconnect();
 
   fConnectString = connect_string;
   fPath = connect_string;
 
   // add trailing '/'
   if (fPath.length() > 0) {
      if (fPath[fPath.length()-1] != DIR_SEPARATOR)
         fPath += DIR_SEPARATOR_STR;
   }
 
   return HS_SUCCESS;
}
 
int FileHistory::hs_clear_cache()
{
   if (fDebug)
      printf("FileHistory::hs_clear_cache!\n");
   fPathLastMtime = 0;
   return SchemaHistoryBase::hs_clear_cache();
}
 
int FileHistory::hs_disconnect()
{
   if (fDebug)
      printf("FileHistory::hs_disconnect!\n");
 
   hs_flush_buffers();
   hs_clear_cache();
 
   return HS_SUCCESS;
}
 
void FileHistory::clear_file_list()
{
   fPathLastMtime = 0;
   fSortedFiles.clear();
   fSortedRead.clear();
}
 
int FileHistory::read_file_list(bool* pchanged)
{
   int status;
   double start_time = ss_time_sec();
 
   if (pchanged)
      *pchanged = false;
 
   struct stat stat_buf;
   status = stat(fPath.c_str(), &stat_buf);
   if (status != 0) {
      cm_msg(MERROR, "FileHistory::read_file_list", "Cannot stat(%s), errno %d (%s)", fPath.c_str(), errno, strerror(errno));
      return HS_FILE_ERROR;
   }
 
   //printf("dir [%s], mtime: %d %d last: %d %d, mtime %s", fPath.c_str(), stat_buf.st_mtimespec.tv_sec, stat_buf.st_mtimespec.tv_nsec, last_mtimespec.tv_sec, last_mtimespec.tv_nsec, ctime(&stat_buf.st_mtimespec.tv_sec));
 
   if (stat_buf.st_mtime == fPathLastMtime) {
      if (fDebug)
         printf("FileHistory::read_file_list: history directory \"%s\" mtime %d did not change\n", fPath.c_str(), int(stat_buf.st_mtime));
      return HS_SUCCESS;
   }
 
   fPathLastMtime = stat_buf.st_mtime;
 
   if (fDebug)
      printf("FileHistory::read_file_list: reading list of history files in \"%s\"\n", fPath.c_str());
 
   std::vector<std::string> flist;
 
   ss_file_find(fPath.c_str(), "mhf_*.dat", &flist);
 
   double ls_time = ss_time_sec();
   double ls_elapsed = ls_time - start_time;
   if (ls_elapsed > 5.000) {
      cm_msg(MINFO, "FileHistory::read_file_list", "\"ls -l\" of \"%s\" took %.1f sec", fPath.c_str(), ls_elapsed);
      cm_msg_flush_buffer();
   }
 
   // note: reverse iterator is used to sort filenames by time, newest first
   std::sort(flist.rbegin(), flist.rend());
 
#if 0
   {
      printf("file names sorted by time:\n");
      for (size_t i=0; i<flist.size(); i++) {
         printf("%d: %s\n", i, flist[i].c_str());
      }
   }
#endif
 
   std::vector<bool> fread;
   fread.resize(flist.size()); // fill with "false"
 
   // loop over the old list of files,
   // for files we already read, loop over new file
   // list and mark the same file as read. K.O.
   for (size_t j=0; j<fSortedFiles.size(); j++) {
      if (fSortedRead[j]) {
         for (size_t i=0; i<flist.size(); i++) {
            if (flist[i] == fSortedFiles[j]) {
               fread[i] = true;
               break;
            }
         }
      }
   }
 
   fSortedFiles = flist;
   fSortedRead  = fread;
 
   if (pchanged)
      *pchanged = true;
 
   return HS_SUCCESS;
}
 
int FileHistory::read_schema(HsSchemaVector* sv, const char* event_name, const time_t timestamp)
{
   if (fDebug)
      printf("FileHistory::read_schema: event [%s] at time %s\n", event_name, TimeToString(timestamp).c_str());
 
   if (sv->size() == 0) {
      if (fDebug)
         printf("FileHistory::read_schema: schema is empty, do a full reload from disk\n");
      clear_file_list();
   }
 
   BOOL old_call_watchdog = FALSE;
   DWORD old_timeout = 0;
   cm_get_watchdog_params(&old_call_watchdog, &old_timeout);
   cm_set_watchdog_params(old_call_watchdog, 0);
 
   bool changed = false;
 
   int status = read_file_list(&changed);
 
   if (status != HS_SUCCESS) {
      cm_set_watchdog_params(old_call_watchdog, old_timeout);
      return status;
   }
 
   if (!changed) {
      if ((*sv).find_event(event_name, timestamp)) {
         if (fDebug)
            printf("FileHistory::read_schema: event [%s] at time %s, no new history files, already have this schema\n", event_name, TimeToString(timestamp).c_str());
         cm_set_watchdog_params(old_call_watchdog, old_timeout);
         return HS_SUCCESS;
      }
   }
 
   double start_time = ss_time_sec();
 
   int count_read = 0;
 
   for (size_t i=0; i<fSortedFiles.size(); i++) {
      std::string file_name = fPath + fSortedFiles[i];
      if (fSortedRead[i])
         continue;
      //bool dupe = false;
      //for (size_t ss=0; ss<sv->size(); ss++) {
      //   HsFileSchema* ssp = (HsFileSchema*)(*sv)[ss];
      //   if (file_name == ssp->fFileName) {
      //      dupe = true;
      //      break;
      //   }
      //}
      //if (dupe)
      //   continue;
      fSortedRead[i] = true;
      HsFileSchema* s = read_file_schema(file_name.c_str());
      if (!s)
         continue;
      sv->add(s);
      count_read++;
 
      if (event_name) {
         if (s->fEventName == event_name) {
            //printf("file %s event_name %s time %s, age %f\n", file_name.c_str(), s->fEventName.c_str(), TimeToString(s->fTimeFrom).c_str(), double(timestamp - s->fTimeFrom));
            if (s->fTimeFrom <= timestamp) {
               // this file is older than the time requested,
               // subsequent files will be even older,
               // we can stop reading here.
               break;
            }
         }
      }
   }
 
   double end_time = ss_time_sec();
   double read_elapsed = end_time - start_time;
   if (read_elapsed > 5.000) {
      cm_msg(MINFO, "FileHistory::read_schema", "Loading schema for event \"%s\" timestamp %s, reading %d history files took %.1f sec", event_name, TimeToString(timestamp).c_str(), count_read, read_elapsed);
      cm_msg_flush_buffer();
   }
 
   cm_set_watchdog_params(old_call_watchdog, old_timeout);
 
   return HS_SUCCESS;
}
 
void FileHistory::tags_to_variables(int ntags, const TAG tags[], std::vector<HsSchemaEntry>& variables)
{
   for (int i=0; i<ntags; i++) {
      HsSchemaEntry e;
 
      int tsize = rpc_tid_size(tags[i].type);
 
      e.tag_name = tags[i].name;
      e.tag_type = rpc_tid_name(tags[i].type);
      e.name     = tags[i].name;
      e.type     = tags[i].type;
      e.n_data   = tags[i].n_data;
      e.n_bytes  = tags[i].n_data*tsize;
 
      variables.push_back(e);
   }
}
 
HsSchema* FileHistory::new_event(const char* event_name, time_t timestamp, int ntags, const TAG tags[])
{
   if (fDebug)
      printf("FileHistory::new_event: event [%s], timestamp %s, ntags %d\n", event_name, TimeToString(timestamp).c_str(), ntags);
 
   int status;
 
   HsFileSchema* s = (HsFileSchema*)fWriterSchema.find_event(event_name, timestamp);
 
   if (!s) {
      //printf("hs_define_event: no schema for event %s\n", event_name);
      status = read_schema(&fWriterSchema, event_name, timestamp);
      if (status != HS_SUCCESS)
         return NULL;
      s = (HsFileSchema*)fWriterSchema.find_event(event_name, timestamp);
   } else {
      //printf("hs_define_event: already have schema for event %s\n", s->fEventName.c_str());
   }
 
   bool xdebug = false;
 
   if (s) { // is existing schema the same as new schema?
      bool same = true;
 
      if (same) {
         if (s->fEventName != event_name) {
            if (xdebug)
               printf("AAA: [%s] [%s]!\n", s->fEventName.c_str(), event_name);
            same = false;
         }
      }
 
      if (same) {
         if (s->fVariables.size() != (size_t)ntags) {
            if (xdebug)
               printf("BBB: event [%s]: ntags: %zu -> %d!\n", event_name, s->fVariables.size(), ntags);
            same = false;
         }
      }
 
      if (same) {
         for (size_t i=0; i<s->fVariables.size(); i++) {
            if (s->fVariables[i].name != tags[i].name) {
               if (xdebug)
                  printf("CCC: event [%s] index %zu: name [%s] -> [%s]!\n", event_name, i, s->fVariables[i].name.c_str(), tags[i].name);
               same = false;
            }
            if (s->fVariables[i].type != (int)tags[i].type) {
               if (xdebug)
                  printf("DDD: event [%s] index %zu: type %d -> %d!\n", event_name, i, s->fVariables[i].type, tags[i].type);
               same = false;
            }
            if (s->fVariables[i].n_data != (int)tags[i].n_data) {
               if (xdebug)
                  printf("EEE: event [%s] index %zu: n_data %d -> %d!\n", event_name, i, s->fVariables[i].n_data, tags[i].n_data);
               same = false;
            }
            if (!same)
               break;
         }
      }
 
      if (!same) {
         if (xdebug) {
            printf("*** Schema for event %s has changed!\n", event_name);
 
            printf("*** Old schema for event [%s] time %s:\n", event_name, TimeToString(timestamp).c_str());
            s->print();
            printf("*** New tags:\n");
            PrintTags(ntags, tags);
         }
 
         if (fDebug)
            printf("FileHistory::new_event: event [%s], timestamp %s, ntags %d: schema mismatch, starting a new file.\n", event_name, TimeToString(timestamp).c_str(), ntags);
 
         s = NULL;
      }
   }
 
   if (s) {
      // maybe this schema is too old - rotate files every so often
      time_t age = timestamp - s->fTimeFrom;
      //printf("*** age %s (%.1f months), timestamp %s, time_from %s, file %s\n", TimeToString(age).c_str(), (double)age/(double)kMonth, TimeToString(timestamp).c_str(), TimeToString(s->fTimeFrom).c_str(), s->fFileName.c_str());
      if (age > fConfMaxFileAge) {
         if (fDebug)
            printf("FileHistory::new_event: event [%s], timestamp %s, ntags %d: schema is too old, age %.1f months, starting a new file.\n", event_name, TimeToString(timestamp).c_str(), ntags, (double)age/(double)kMonth);
 
         // force creation of a new file
         s = NULL;
      }
   }
 
   if (s) {
      // maybe this file is too big - rotate files to limit maximum size
      double size = ss_file_size(s->fFileName.c_str());
      //printf("*** size %.0f, file %s\n", size, s->fFileName.c_str());
      if (size > fConfMaxFileSize) {
         if (fDebug)
            printf("FileHistory::new_event: event [%s], timestamp %s, ntags %d: file too big, size %.1f MiBytes, max size %.1f MiBytes, starting a new file.\n", event_name, TimeToString(timestamp).c_str(), ntags, size/MiB, fConfMaxFileSize/MiB);
 
         // force creation of a new file
         s = NULL;
      }
   }
 
   if (!s) {
      std::string filename;
 
      std::vector<HsSchemaEntry> vars;
 
      tags_to_variables(ntags, tags, vars);
 
      status = create_file(event_name, timestamp, vars, &filename);
      if (status != HS_SUCCESS)
         return NULL;
 
      HsFileSchema* ss = read_file_schema(filename.c_str());
      if (!ss) {
         cm_msg(MERROR, "FileHistory::new_event", "Error: Cannot create schema for event \'%s\', see previous messages", event_name);
         return NULL;
      }
 
      fWriterSchema.add(ss);
 
      s = (HsFileSchema*)fWriterSchema.find_event(event_name, timestamp);
 
      if (!s) {
         cm_msg(MERROR, "FileHistory::new_event", "Error: Cannot create schema for event \'%s\', see previous messages", event_name);
         return NULL;
      }
 
      if (xdebug) {
         printf("*** New schema for event [%s] time %s:\n", event_name, TimeToString(timestamp).c_str());
         s->print();
      }
   }
 
   assert(s != NULL);
 
#if 0
   {
      printf("schema for [%s] is %p\n", event_name, s);
      if (s)
         s->print();
   }
#endif
 
   HsFileSchema* e = new HsFileSchema();
 
   *e = *s; // make a copy of the schema
 
   return e;
}
 
int FileHistory::create_file(const char* event_name, time_t timestamp, const std::vector<HsSchemaEntry>& vars, std::string* filenamep)
{
   if (fDebug)
      printf("FileHistory::create_file: event [%s]\n", event_name);
 
   // NB: file names are constructed in such a way
   // that when sorted lexicographically ("ls -1 | sort")
   // they *also* become sorted by time
 
   struct tm tm;
   localtime_r(&timestamp, &tm); // somebody must call tzset() before this.
 
   char buf[256];
   strftime(buf, sizeof(buf), "%Y%m%d", &tm);
 
   std::string filename;
   filename += fPath;
   filename += "mhf_";
   filename += TimeToString(timestamp);
   filename += "_";
   filename += buf;
   filename += "_";
   filename += MidasNameToFileName(event_name);
 
   std::string try_filename = filename + ".dat";
 
   FILE *fp = NULL;
   for (int itry=0; itry<10; itry++) {
      if (itry > 0) {
         char s[256];
         sprintf(s, "_%d", rand());
         try_filename = filename + s + ".dat";
      }
 
      fp = fopen(try_filename.c_str(), "r");
      if (fp != NULL) {
         // this file already exists, try with a different name
         fclose(fp);
         continue;
      }
 
      fp = fopen(try_filename.c_str(), "w");
      if (fp == NULL) {
         // somehow cannot create this file, try again
         cm_msg(MERROR, "FileHistory::create_file", "Error: Cannot create file \'%s\' for event \'%s\', fopen() errno %d (%s)", try_filename.c_str(), event_name, errno, strerror(errno));
         continue;
      }
 
      // file opened
      break;
   }
 
   if (fp == NULL) {
      // somehow cannot create any file, whine!
      cm_msg(MERROR, "FileHistory::create_file", "Error: Cannot create file \'%s\' for event \'%s\'", filename.c_str(), event_name);
      return HS_FILE_ERROR;
   }
 
   std::string ss;
 
   ss += "version: 2.0\n";
   ss += "event_name: ";
   ss    += event_name;
   ss    += "\n";
   ss += "time: ";
   ss    += TimeToString(timestamp);
   ss    += "\n";
 
   int recsize = 0;
 
   ss += "tag: /DWORD 1 4 /timestamp\n";
   recsize += 4;
 
   bool padded = false;
   int offset = 0;
 
   bool xdebug = false; // (strcmp(event_name, "u_Beam") == 0);
 
   for (size_t i=0; i<vars.size(); i++) {
      int tsize = rpc_tid_size(vars[i].type);
      int n_bytes = vars[i].n_data*tsize;
      int xalign = (offset % tsize);
 
      if (xdebug)
         printf("tag %zu, tsize %d, n_bytes %d, xalign %d\n", i, tsize, n_bytes, xalign);
 
#if 0
      // looks like history data does not do alignement and padding
      if (xalign != 0) {
         padded = true;
         int pad_bytes = tsize - xalign;
         assert(pad_bytes > 0);
 
         ss += "tag: ";
         ss    += "XPAD";
         ss    += " ";
         ss    += SmallIntToString(1);
         ss    += " ";
         ss    += SmallIntToString(pad_bytes);
         ss    += " ";
         ss    += "pad_";
         ss    += SmallIntToString(i);
         ss    += "\n";
 
         offset += pad_bytes;
         recsize += pad_bytes;
 
         assert((offset % tsize) == 0);
         fprintf(stderr, "FIXME: need to debug padding!\n");
         //abort();
      }
#endif
      
      ss += "tag: ";
      ss    += rpc_tid_name(vars[i].type);
      ss    += " ";
      ss    += SmallIntToString(vars[i].n_data);
      ss    += " ";
      ss    += SmallIntToString(n_bytes);
      ss    += " ";
      ss    += vars[i].name;
      ss    += "\n";
 
      recsize += n_bytes;
      offset += n_bytes;
   }
 
   ss += "record_size: ";
   ss    += SmallIntToString(recsize);
   ss    += "\n";
 
   // reserve space for "data_offset: ..."
   int sslength = ss.length() + 127;
 
   int block = 1024;
   int nb = (sslength + block - 1)/block;
   int data_offset = block * nb;
 
   ss += "data_offset: ";
   ss    += SmallIntToString(data_offset);
   ss    += "\n";
 
   fprintf(fp, "%s", ss.c_str());
 
   fclose(fp);
 
   if (1 && padded) {
      printf("Schema in file %s has padding:\n", try_filename.c_str());
      printf("%s", ss.c_str());
   }
 
   if (filenamep)
      *filenamep = try_filename;
 
   return HS_SUCCESS;
}
 
HsFileSchema* FileHistory::read_file_schema(const char* filename)
{
   if (fDebug)
      printf("FileHistory::read_file_schema: file %s\n", filename);
 
   FILE* fp = fopen(filename, "r");
   if (!fp) {
      cm_msg(MERROR, "FileHistory::read_file_schema", "Cannot read \'%s\', fopen() errno %d (%s)", filename, errno, strerror(errno));
      return NULL;
   }
 
   HsFileSchema* s = NULL;
 
   // File format looks like this:
   // version: 2.0
   // event_name: u_Beam
   // time: 1023174012
   // tag: /DWORD 1 4 /timestamp
   // tag: FLOAT 1 4 B1
   // ...
   // tag: FLOAT 1 4 Ref Heater
   // record_size: 84
   // data_offset: 1024
 
   size_t rd_recsize = 0;
   int offset = 0;
 
   while (1) {
      char buf[1024];
      char* b = fgets(buf, sizeof(buf), fp);
 
      //printf("read: %s\n", b);
 
      if (!b) {
         break; // end of file
      }
 
      char*bb;
 
      bb = strchr(b, '\n');
      if (bb)
         *bb = 0;
 
      bb = strchr(b, '\r');
      if (bb)
         *bb = 0;
 
      bb = strstr(b, "version: 2.0");
      if (bb == b) {
         s = new HsFileSchema();
         assert(s);
 
         s->fFileName = filename;
         continue;
      }
 
      if (!s) {
         // malformed history file
         break;
      }
 
      bb = strstr(b, "event_name: ");
      if (bb == b) {
         s->fEventName = bb + 12;
         continue;
      }
 
      bb = strstr(b, "time: ");
      if (bb == b) {
         s->fTimeFrom = strtoul(bb + 6, NULL, 10);
         continue;
      }
 
      // tag format is like this:
      //
      // tag: FLOAT 1 4 Ref Heater
      //
      // "FLOAT" is the MIDAS type, "/DWORD" is special tag for the timestamp
      // "1" is the number of array elements
      // "4" is the total tag size in bytes (n_data*tid_size)
      // "Ref Heater" is the tag name
 
      bb = strstr(b, "tag: ");
      if (bb == b) {
         bb += 5; // now points to the tag MIDAS type
         const char* midas_type = bb;
         char* bbb = strchr(bb, ' ');
         if (bbb) {
            *bbb = 0;
            HsSchemaEntry t;
            if (midas_type[0] == '/') {
               t.type = 0;
            } else {
               t.type = rpc_name_tid(midas_type);
               if (t.type == 0) {
                  cm_msg(MERROR, "FileHistory::read_file_schema", "Unknown MIDAS data type \'%s\' in history file \'%s\'", midas_type, filename);
                  if (s)
                     delete s;
                  s = NULL;
                  break;
               }
            }
            bbb++;
            while (*bbb == ' ')
               bbb++;
            if (*bbb) {
               t.n_data = strtoul(bbb, &bbb, 10);
               while (*bbb == ' ')
                  bbb++;
               if (*bbb) {
                  t.n_bytes = strtoul(bbb, &bbb, 10);
                  while (*bbb == ' ')
                     bbb++;
                  t.name = bbb;
               }
            }
 
            if (midas_type[0] != '/') {
               s->fVariables.push_back(t);
               s->fOffsets.push_back(offset);
               offset += t.n_bytes;
            }
 
            rd_recsize += t.n_bytes;
         }
         continue;
      }
 
      bb = strstr(b, "record_size: ");
      if (bb == b) {
         s->fRecordSize = atoi(bb + 12);
         continue;
      }
 
      bb = strstr(b, "data_offset: ");
      if (bb == b) {
         s->fDataOffset = atoi(bb + 12);
         // data offset is the last entry in the file
         break;
      }
   }
 
   fclose(fp);
 
   if (!s) {
      cm_msg(MERROR, "FileHistory::read_file_schema", "Malformed history schema in \'%s\', maybe it is not a history file", filename);
      return NULL;
   }
 
   if (rd_recsize != s->fRecordSize) {
      cm_msg(MERROR, "FileHistory::read_file_schema", "Record size mismatch in history schema from \'%s\', file says %zu while total of all tags is %zu", filename, s->fRecordSize, rd_recsize);
      if (s)
         delete s;
      return NULL;
   }
 
   if (!s) {
      cm_msg(MERROR, "FileHistory::read_file_schema", "Could not read history schema from \'%s\', maybe it is not a history file", filename);
      if (s)
         delete s;
      return NULL;
   }
 
   if (fDebug > 1)
      s->print();
 
   return s;
}
 
HsSchema* FileHistory::maybe_reopen(const char* event_name, time_t timestamp, HsSchema* s)
{
   HsFileSchema* fs = dynamic_cast<HsFileSchema*>(s);
 
   assert(fs != NULL); // FileHistory::maybe_reopen() must be called only with file history schema.
 
   if (fs->fFileSize <= fConfMaxFileSize) {
      // not big enough, let it grow
      return s;
   }
 
   // must rotate the file
 
   if (fDebug) {
#ifdef OS_DARWIN
      printf("FileHistory::maybe_reopen: reopen file \"%s\", size %lld, max size %lld\n", fs->fFileName.c_str(), fs->fFileSize, fConfMaxFileSize);
#else
      printf("FileHistory::maybe_reopen: reopen file \"%s\", size %jd, max size %jd\n", fs->fFileName.c_str(), fs->fFileSize, fConfMaxFileSize);
#endif
   }
   
   std::string new_filename;
   
   int status = create_file(event_name, timestamp, fs->fVariables, &new_filename);
   
   if (status != HS_SUCCESS) {
      // cannot create new history file
      return s;
   }
   
   HsFileSchema* new_fs = read_file_schema(new_filename.c_str());
   
   if (!new_fs) {
      // cannot open new history file
      return s;
   }
 
   new_fs->fDisabled = false;
 
   HsFileSchema* new_fs_copy = new HsFileSchema;
   *new_fs_copy = *new_fs; // make a copy
 
   //printf("replacing schema %p %p with %p %p\n", s, fs, new_fs, new_fs_copy);
   
   for (size_t i=0; i<fWriterEvents.size(); i++) {
      if (s == fWriterEvents[i]) {
         delete fWriterEvents[i];
         fWriterEvents[i] = new_fs;
         s = NULL; // pointer to fEvents[i]
         fs = NULL; // pointer to fEvents[i]
      }
   }
   
   assert(s == NULL); // the schema we are replacing must be in fWriterEvents.
 
   fWriterSchema.add(new_fs_copy); // make sure new file is added to the list of files
   
   assert(new_fs->fFileSize < fConfMaxFileSize); // check that we are not returning the original big file
 
   //new_fs->print();
   
   return new_fs;
}
 
//               Factory constructors                 //
 
MidasHistoryInterface* MakeMidasHistorySqlite()
{
#ifdef HAVE_SQLITE
   return new SqliteHistory();
#else
   cm_msg(MERROR, "MakeMidasHistorySqlite", "Error: Cannot initialize SQLITE history - this MIDAS was built without SQLITE support - HAVE_SQLITE is not defined");
   return NULL;
#endif
}
 
MidasHistoryInterface* MakeMidasHistoryMysql()
{
#ifdef HAVE_MYSQL
   return new MysqlHistory();
#else
   cm_msg(MERROR, "MakeMidasHistoryMysql", "Error: Cannot initialize MySQL history - this MIDAS was built without MySQL support - HAVE_MYSQL is not defined");
   return NULL;
#endif
}
 
MidasHistoryInterface* MakeMidasHistoryPgsql()
{
#ifdef HAVE_PGSQL
   return new PgsqlHistory();
#else
   cm_msg(MERROR, "MakeMidasHistoryPgsql", "Error: Cannot initialize PgSQL history - this MIDAS was built without PostgreSQL support - HAVE_PGSQL is not defined");
   return NULL;
#endif
}
 
MidasHistoryInterface* MakeMidasHistoryFile()
{
   return new FileHistory();
}
 
/* emacs
 * Local Variables:
 * tab-width: 8
 * c-basic-offset: 3
 * indent-tabs-mode: nil
 * End:
 */