DBFolder.cxx

Go to the documentation of this file.

#include "DBFolder.h"
#include "WebDBIConstants.h"
#include "larevt/CalibrationDBI/IOVData/TimeStampDecoder.h"
#include "WebError.h"

#include <sstream>
#include <stdlib.h>
#include <cstring>
#include "wda.h"
#include "sqlite3.h"
#include "cetlib_except/exception.h"
#include "cetlib/search_path.h"
#include "messagefacility/MessageLogger/MessageLogger.h"

namespace lariov {

  // Constructor.

  DBFolder::DBFolder(const std::string& name, const std::string& url, const std::string& url2,
                     const std::string& tag, bool usesqlite, bool testmode)
  {
    fFolderName = name;
    fURL = url;
    fURL2 = url2;
    fTag = tag;
    fUseSQLite = usesqlite;
    fTestMode = testmode;
    if (fURL[fURL.length()-1] == '/') {
      fURL = fURL.substr(0, fURL.length()-1);
    }

    fCachedRowNumber = -1;
    fCachedChannel = 0;

    fMaximumTimeout = 4*60; //4 minutes

    // If UsqSQLite is true, hunt for sqlite database file.
    // It is an error if this file can't be found.

    //mf::LogInfo("DBFolder") << "DBFolder: Folder name = " << fFolderName << "\n";
    if(fUseSQLite) {
      std::string dbname = fFolderName + ".db";
      cet::search_path sp("FW_SEARCH_PATH");
      fSQLitePath = sp.find_file(dbname);   // Throws exception if not found.
      //mf::LogInfo("DBFolder") << "DBFolder: SQLite database path = " << fSQLitePath << "\n";
    }
    //else
    //  mf::LogInfo("DBFolder") << "DBFolder: database url = " << fURL << "\n";

    if(fTestMode && fURL2 != "") {
      mf::LogInfo log("DBFolder");
      log << "\nDBFolder test mode, will compare the following urls data." << "\n";
      log << fURL << "\n";
      log << fURL2 << "\n" << "\n";
    }
    if(fTestMode && fUseSQLite) {
      mf::LogInfo log("DBFolder");
      log << "\nDBFolder test mode, will compare the following url and sqlite data." << "\n";
      log << fURL << "\n";
      log << fSQLitePath << "\n" << "\n";
    }
  }

  // Destructor.

  DBFolder::~DBFolder() {}

  // Data accessors.

  int DBFolder::GetNamedChannelData(DBChannelID_t channel, const std::string& name, bool& data) {

    int err = 0;

    // Make sure cached row is valid.

    GetRow(channel);

    // Get column index.

    size_t col = GetColumn(name);

    // Get value.

    long value = fCachedRow.getLongData(col);
    data = (value != 0);

    return err;
  }

  int DBFolder::GetNamedChannelData(DBChannelID_t channel, const std::string& name, long& data) {

    int err = 0;

    // Make sure cached row is valid.

    GetRow(channel);

    // Get column index.

    size_t col = GetColumn(name);

    // Get value.

    data = fCachedRow.getLongData(col);

    // Done.

    return err;
  }

  int DBFolder::GetNamedChannelData(DBChannelID_t channel, const std::string& name, double& data) {

    int err = 0;

    // Make sure cached row is valid.

    GetRow(channel);

    // Get column index.

    size_t col = GetColumn(name);

    // Get value.

    data = fCachedRow.getDoubleData(col);

    // Done.

    return err;
  }

  int DBFolder::GetNamedChannelData(DBChannelID_t channel, const std::string& name, std::string& data) {

    int err = 0;

    // Make sure cached row is valid.

    GetRow(channel);

    // Get column index.

    size_t col = GetColumn(name);

    // Get value.

    data = fCachedRow.getStringData(col);

    // Done.

    return err;
  }

  // Not sure why the following accessor is included.  Doesn't seem to be used.

  /*
  int DBFolder::GetNamedChannelData(DBChannelID_t channel, const std::string& name, std::vector<double>& data) {

    data.clear();

    Tuple tup;
    size_t col = this->GetRow(channel, name, tup);
    int err=0;
    double buf[kBUFFER_SIZE];

    DataRec *dataRec = (DataRec *)tup;
    // for c2: col is an unsigned int and cannot be less than 0
    // if (col < 0 || col >= dataRec->ncolumns) {
    if (col >= dataRec->ncolumns) {
      err=-1;
      return err;
    }

    char* sptr = dataRec->columns[col];
    if ( *sptr == '[') sptr +=1;  //expect an initial bracket and skip it
    else {
      err=-2;
      return err;
    }

    char* eptr;
    double val;
    unsigned int array_size=0;
    for (unsigned int i=0; i < kBUFFER_SIZE; ++i) {
      val = strtod(sptr, &eptr); //Try to convert
      if (sptr==eptr) break;     //conversion failed
      if (*sptr=='\0') break;    //end loop if buffer ends

      buf[array_size++] = val;

      if ( *eptr == ']') break;  //found the closing bracket, we're done
      else sptr = eptr+1;        //point to the next value
    }

    data.insert(data.begin(), buf, buf + array_size);
    releaseTuple(tup);
    return err;
  }
  */

  int DBFolder::GetChannelList( std::vector<DBChannelID_t>& channels ) const {

    channels = fCache.channels();
    return 0;
  }

  // Update cached row.

  void DBFolder::GetRow(DBChannelID_t channel) {

    // Check if we need to update the cached row.

    if (fCachedChannel != channel ||
        !fCachedRow.isValid() ||
        fCachedRow.getLongData(0) != channel) {

      // Update cached row number (binary serach).

      int row = fCache.getRowNumber(channel);

      //  Throw an exception if we didn't find a matching role.

      if(row < 0) {
        std::string msg = "Channel " + std::to_string(channel) + " is not found in database!";
        throw WebError(msg);
      }

      // Update cached row.

      fCachedRowNumber = row;
      fCachedChannel = channel;
      fCachedRow = fCache.getRow(row);
    }
  }

  // Find matching column.

  size_t DBFolder::GetColumn(const std::string& name) const
  {
    int col = fCache.getColNumber(name);

    // See if we found a matching column.

    if(col < 0) {
      std::string msg = "Column " + name + " is not found in database!";
      throw WebError(msg);
    }

    // Done.

    return col;
  }

  //returns true if an Update is performed, false if not
  bool DBFolder::UpdateData( DBTimeStamp_t raw_time) {

    //convert to IOVTimeStamp
    IOVTimeStamp ts = TimeStampDecoder::DecodeTimeStamp(raw_time);

    //check if cache is updated
    if (IsValid(ts)) return false;

    //release cached data.
    fCache = DBDataset();
    fCachedRow = DBDataset::DBRow();
    fCachedRowNumber = -1;
    fCachedChannel = 0;

    //get full url string
    std::stringstream fullurl;
    fullurl << fURL << "/data?f=" << fFolderName
            << "&t=" << ts.DBStamp();
    if (fTag.length() > 0) fullurl << "&tag=" << fTag;

    //mf::LogInfo log("DBFolder")
    //log << "In DBFolder::UpdateData" << "\n";
    //log << "t=" << raw_time/1000000000 << "\n";
    //log << "Full url = " << fullurl.str() << "\n";

    //get new dataset
    if(fSQLitePath != "" && !fTestMode) {
      GetSQLiteData(raw_time/1000000000, fCache);
    }
    else {
      if(fTestMode) {
        mf::LogInfo log("DBFolder");
        log << "Accessing primary calibration data from http conditions database server." << "\n";
        log << "Folder = " << fFolderName << "\n";
      }
      int err = 0;
      Dataset data = getDataWithTimeout(fullurl.str().c_str(), NULL, fMaximumTimeout, &err);
      int status = getHTTPstatus(data);
      if (status != 200) {
        std::string msg = "HTTP error from " + fullurl.str()+": status: " + std::to_string(status) + ": " + std::string(getHTTPmessage(data));
        throw WebError(msg);
      }
      fCache = DBDataset(data, true);
    }
    //DumpDataset(fCache);


    // If test mode is selected, get comparison data.

    if(fTestMode) {
      if(fSQLitePath != "") {
        DBDataset compare1;
        mf::LogInfo("DBFolder") << "Accessing comparison data from sqlite database " << fSQLitePath << "\n";
        GetSQLiteData(raw_time/1000000000, compare1);
        CompareDataset(fCache, compare1);
      }
      if(fURL2 != "") {
        mf::LogInfo("DBFolder") <<"Accessing comparison data from second database url." << "\n";
        std::stringstream fullurl2;
        fullurl2 << fURL2 << "/data?f=" << fFolderName
                 << "&t=" << ts.DBStamp();
        if (fTag.length() > 0) fullurl2 << "&tag=" << fTag;
        mf::LogInfo("DBFolder") << "Full url = " << fullurl2.str() << "\n";
        int err = 0;
        Dataset data = getDataWithTimeout(fullurl2.str().c_str(), NULL, fMaximumTimeout, &err);
        int status = getHTTPstatus(data);
        if (status != 200) {
          std::string msg = "HTTP error from " + fullurl2.str()+": status: " + std::to_string(status) + ": " + std::string(getHTTPmessage(data));
          throw WebError(msg);
        }
        DBDataset compare2(data, true);
        CompareDataset(fCache, compare2);
      }
    }
    return true;
  }

  // Query data from sqlite database.
  // The return value of type Dataset (aka void*), is partially opaque type HttpResponse*
  // (defined in wda.c and copied above).

  void DBFolder::GetSQLiteData(int t, DBDataset& data) const
  {
    if(fSQLitePath == "")
      return;

    // DBDataset data to be filled.

    IOVTimeStamp begin_ts(0, 0);                // IOV begin time.
    IOVTimeStamp end_ts(0, 0);                  // IOV end time.
    std::vector<std::string> column_names;      // Column names.
    std::vector<std::string> column_types;      // Column types.
    std::vector<DBChannelID_t> channels;        // Channels.
    std::vector<DBDataset::value_type> values;  // Calibration data (length nchan*ncols).

    //mf::LogInfo log("DBFolder")
    //log << "DBFolder::GetSQLiteData" << "\n";
    //log << "t=" << t << "\n";
    //log << "sqlite path = " << fSQLitePath << "\n";

    // Open sqlite database.

    //mf::LogInfo("DBFolder") << "Opening sqlite database " << fSQLitePath << "\n";
    sqlite3* db;
    int rc = sqlite3_open(fSQLitePath.c_str(), &db);
    if(rc != SQLITE_OK) {
      mf::LogError("DBFolder") << "Failed to open sqlite database " << fSQLitePath << "\n";
      throw cet::exception("DBFolder") << "Failed to open sqlite database " << fSQLitePath;
    }

    // Query begin time of IOV.

    std::string table_iovs = fFolderName + "_iovs";
    std::string table_tag_iovs = fFolderName + "_tag_iovs";
    std::ostringstream sql;
    sql << "SELECT " << table_iovs << ".iov_id," << table_iovs << ".begin_time"
        << " FROM " << table_tag_iovs << "," << table_iovs
        << " WHERE " << table_tag_iovs << ".tag='" << fTag << "'"
        << " AND " << table_tag_iovs << ".iov_id=" << table_iovs << ".iov_id"
        << " AND " << table_iovs << ".begin_time <= " << t
        << " ORDER BY " << table_iovs << ".begin_time desc";
    //mf::LogInfo("DBFolder") << "sql = " << sql.str() << "\n";

    // Prepare query.

    sqlite3_stmt* stmt;
    rc = sqlite3_prepare_v2(db, sql.str().c_str(), -1, &stmt, 0);
    if(rc != SQLITE_OK) {
      mf::LogError log("DBFolder");
      log << "sqlite3_prepare_v2 failed." << fSQLitePath << "\n";
      log << "Failed sql = " << sql.str() << "\n";
      throw cet::exception("DBFolder") << "sqlite3_prepare_v2 error.";
    }

    // Execute query.
    // Just retrieve first row.
    // It is an error if we don't get at least one row.

    rc = sqlite3_step(stmt);
    //int iov_id = 0;
    int begin_time = 0;
    if(rc == SQLITE_ROW) {
      //iov_id = sqlite3_column_int(stmt, 0);
      begin_time = sqlite3_column_int(stmt, 1);
      //mf::LogInfo log("DBFolder")
      //log << "iov_id = " << iov_id << "\n";
      //log << "begin_time = " << begin_time << "\n";
    }
    else {
      mf::LogError("DBFolder") << "sqlite3_step returned error result = " << rc << "\n";
      throw cet::exception("DBFolder") << "sqlite3_step error.";
    }

    // Delete query.

    sqlite3_finalize(stmt);

    // Query end time of IOV.

    sql.str("");
    sql << "SELECT " << table_iovs << ".begin_time"
        << " FROM " << table_tag_iovs << "," << table_iovs
        << " WHERE " << table_tag_iovs << ".tag='" << fTag << "'"
        << " AND " << table_tag_iovs << ".iov_id=" << table_iovs << ".iov_id"
        << " AND " << table_iovs << ".begin_time > " << t
        << " ORDER BY " << table_iovs << ".begin_time";
    //mf::LogInfo("DBFolder") << "sql = " << sql.str() << "\n";

    // Prepare query.

    rc = sqlite3_prepare_v2(db, sql.str().c_str(), -1, &stmt, 0);
    if(rc != SQLITE_OK) {
      mf::LogError log("DBFolder");
      log << "sqlite3_prepare_v2 failed." << fSQLitePath << "\n";
      log << "Failed sql = " << sql.str() << "\n";
      throw cet::exception("DBFolder") << "sqlite3_prepare_v2 error.";
    }

    // Execute query.
    // Just retrieve first row.
    // If we don't get any rows, then end time is infinite.

    rc = sqlite3_step(stmt);
    int end_time = 0;
    if(rc == SQLITE_ROW) {
      end_time = sqlite3_column_int(stmt, 0);
      //mf::LogInfo("DBFolder") << "end_time = " << end_time << "\n";
    }
    else if(rc != SQLITE_DONE) {
      mf::LogError("DBFolder") << "sqlite3_step returned error result = " << rc << "\n";
      throw cet::exception("DBFolder") << "sqlite3_step error.";
    }

    // Delete query.

    sqlite3_finalize(stmt);

    // Query count of channels.
    // We do this so that we know how much memory to allocate.

    std::string table_data = fFolderName + "_data";
    sql.str("");
    sql << "SELECT COUNT(DISTINCT channel)"
        << " FROM " << table_data << "," << table_iovs << "," << table_tag_iovs
        << " WHERE " << table_tag_iovs << ".tag='" << fTag << "'"
        << " AND " << table_iovs << ".iov_id=" << table_tag_iovs << ".iov_id"
        << " AND " << table_data << ".__iov_id=" << table_tag_iovs << ".iov_id"
        << " AND " << table_iovs << ".begin_time <= " << t;
    //mf::LogInfo("DBFolder") << "sql = " << sql.str() << "\n";

    // Prepare query.

    rc = sqlite3_prepare_v2(db, sql.str().c_str(), -1, &stmt, 0);
    if(rc != SQLITE_OK) {
      mf::LogError log("DBFolder");
      log << "sqlite3_prepare_v2 failed." << fSQLitePath << "\n";
      log << "Failed sql = " << sql.str() << "\n";
      throw cet::exception("DBFolder") << "sqlite3_prepare_v2 error.";
    }

    // Execute query.
    // Retrieve one row.
    // It is an error if we don't get at least one row.

    rc = sqlite3_step(stmt);
    unsigned int nrows = 0;
    if(rc == SQLITE_ROW) {
      nrows = sqlite3_column_int(stmt, 0);
      //mf::LogInfo("DBFolder") << "Number of data rows = " << nrows << "\n";
    }
    else {
      mf::LogError("DBFolder") << "sqlite3_step returned error result = " << rc << "\n";
      throw cet::exception("DBFolder") << "sqlite3_step error.";
    }

    // Reserve collections that depend on number of rows (only).

    channels.reserve(nrows);

    // Delete query.

    sqlite3_finalize(stmt);

    // Stash begin time.

    begin_ts = IOVTimeStamp(begin_time, 0);

    // Stash end time.

    if(end_time == 0)
      end_ts = IOVTimeStamp::MaxTimeStamp();
    else
      end_ts = IOVTimeStamp(end_time, 0);

    // Main data query.

    sql.str("");
    sql << "SELECT " << table_data << ".*,MAX(begin_time)"
        << " FROM " << table_data << "," << table_iovs << "," << table_tag_iovs
        << " WHERE " << table_tag_iovs << ".tag='" << fTag << "'"
        << " AND " << table_iovs << ".iov_id=" << table_tag_iovs << ".iov_id"
        << " AND " << table_data << ".__iov_id=" << table_tag_iovs << ".iov_id"
        << " AND " << table_iovs << ".begin_time <= " << t
        << " GROUP BY channel"
        << " ORDER BY channel";
    //mf::LogInfo("DBFolder") << "sql = " << sql.str() << "\n";

    // Prepare query.

    rc = sqlite3_prepare_v2(db, sql.str().c_str(), -1, &stmt, 0);
    if(rc != SQLITE_OK) {
      mf::LogError log("DBFolder");
      log << "sqlite3_prepare_v2 failed." << fSQLitePath << "\n";
      log << "Failed sql = " << sql.str() << "\n";
      throw cet::exception("DBFolder") << "sqlite3_prepare_v2 error.";
    }

    // Execute the query and retrieve one row.
    // We do this to extract the number, names, and types of relevant columns.

    int ncols = sqlite3_column_count(stmt);
    column_names.reserve(ncols);
    column_types.reserve(ncols);

    rc = sqlite3_step(stmt);
    if(rc == SQLITE_ROW) {

      // Loop over columns.

      for(int col = 0; col < ncols; ++col) {
        std::string colname = sqlite3_column_name(stmt, col);

        // Ignore columns that begin with "_".
        // Also ignore utility column MAX(begin_time).

        if(colname[0] != '_' && colname.substr(0,3) != "MAX") {
          column_names.push_back(colname);
          int dtype = sqlite3_column_type(stmt, col);
          if(dtype == SQLITE_INTEGER)
            column_types.push_back("integer");
          else if(dtype == SQLITE_FLOAT)
            column_types.push_back("real");
          else if(dtype == SQLITE_TEXT)
            column_types.push_back("text");
          else if(dtype == SQLITE_NULL)
            column_types.push_back("NULL");
          else {
            mf::LogError("DBFolder") << "Unknown type " << dtype << "\n";
            throw cet::exception("DBFolder") << "Unknown type " << dtype;
          }
          //mf::LogInfo("DBFolder") << "Column " << col
          //        << ", name=" << column_names.back()
          //        << ", type=" << column_types.back() << "\n";
        }
      }
    }
    else {
      mf::LogError("DBFolder") << "No data rows." << "\n";
      throw cet::exception("DBFolder") << "No data rows.";
    }

    // Remember the number of relevant columns.

    size_t nrelcols = column_names.size();
    values.reserve(nrows * nrelcols);

    // Re-execute query.
    // Retrieve all data rows and stash in result.

    rc = sqlite3_reset(stmt);
    if(rc != SQLITE_OK) {
      mf::LogError("DBFolder") << "sqlite3_reset failed." << "\n";
      throw cet::exception("DBFolder") << "sqlite3_failed.";
    }
    size_t irow = 0;
    while(rc != SQLITE_DONE) {
      rc = sqlite3_step(stmt);
      if(rc == SQLITE_ROW) {
        ++irow;
        //mf::LogInfo("DBFolder") << irow << " rows." << "\n";
        if(irow > nrows) {
          mf::LogError("DBFolder") << "Too many data rows " << irow << "\n";
          throw cet::exception("DBFolder") << "Too many data rows " << irow;
        }

        // Loop over columns.
        // Remember that ncols is the number of columns returned by the query,
        // not the number of columns that get stored (some columns are ignored).

        bool firstcol = true;
        for(int col = 0; col < ncols; ++col) {
          std::string colname = sqlite3_column_name(stmt, col);

          // Ignore columns that begin with "_".
          // Also ignore utility column MAX(begin_time).

          if(colname[0] != '_' && colname.substr(0,3) != "MAX") {
            int dtype = sqlite3_column_type(stmt, col);

            if(dtype == SQLITE_INTEGER) {
              long value = sqlite3_column_int(stmt, col);
              //mf::LogInfo("DBFolder") << "Value = " << value << "\n";
              values.push_back(DBDataset::value_type(value));
              if(firstcol)
                channels.push_back(value);
            }
            else if(dtype == SQLITE_FLOAT) {
              double value = sqlite3_column_double(stmt, col);
              //mf::LogInfo("DBFolder") << "Value = " << value << "\n";
              values.push_back(DBDataset::value_type(value));
              if(firstcol) {
                mf::LogError("DBFolder") << "First column has wrong type float." << "\n";
                throw cet::exception("DBFolder") << "First column has wrong type float.";
              }
            }
            else if(dtype == SQLITE_TEXT) {
              const char* s = (const char*)sqlite3_column_text(stmt, col);
              //mf::LogInfo("DBFolder") << "Value = " << s << "\n";
              values.emplace_back(std::make_unique<std::string>(s));
              if(firstcol) {
                mf::LogError("DBFolder") << "First column has wrong type text." << "\n";
                throw cet::exception("DBFolder") << "First column has wrong type text.";
              }
            }
            else if(dtype == SQLITE_NULL) {
              values.push_back(DBDataset::value_type());
              //mf::LogInfo("DBFolder") << "Value = NULL" << "\n";
              if(firstcol) {
                mf::LogError("DBFolder") << "First column has wrong type null." << "\n";
                throw cet::exception("DBFolder") << "First column has wrong type null.";
              }
            }
            else {
              mf::LogError("DBFolder") << "Unrecognized sqlite data type" << "\n";
              throw cet::exception("DBFolder") << "Unrecognized sqlite data type.";
            }
            firstcol = false;
          }
        }
      }
      else if(rc != SQLITE_DONE) {
        mf::LogError("DBFolder") << "sqlite3_step returned error result = " << rc << "\n";
        throw cet::exception("DBFolder") << "sqlite3_step error.";
      }
    }
    if(irow != nrows) {
      mf::LogError("DBFolder") << "Wrong number of data rows " << irow << "," << nrows << "\n";
      throw cet::exception("DBFolder") << "Wrong number of data rows " << irow << "," << nrows << "\n";
    }
    if(values.size() != nrows * nrelcols) {
      mf::LogError("DBFolder") << "Wrong number of values "
                               << values.size() << "," << nrows << "," << nrelcols << "\n";
      throw cet::exception("DBFolder") << "Wrong number of values "
                                       << values.size() << "," << nrows << "," << nrelcols << "\n";
    }

    // Delete statement.

    sqlite3_finalize(stmt);

    // Close database.

    sqlite3_close(db);

    // Fill result.

    data = DBDataset(begin_ts, end_ts,
                     std::move(column_names),
                     std::move(column_types),
                     std::move(channels),
                     std::move(values));

    // Done.

    return;
  }

  // Dump dataset by rows.

  void DBFolder::DumpDataset(const DBDataset& data) const
  {
    size_t nrows = data.nrows();
    size_t ncols = data.ncols();
    mf::LogInfo log("DBFolder");
    log << "Dataset contains " << nrows << " rows and " << ncols << " columns." << "\n";

    // Begin time.

    log << "IOV start time = " << data.beginTime().DBStamp() << "\n";

    // End time.

    log << "IOV end time = " << data.endTime().DBStamp() << "\n";

    // Columnn names.

    const std::vector<std::string>& names = data.colNames();
    for (size_t c=0; c<ncols; ++c)
      log << "Column " << c << ", name = " << names[c] << "\n";

    // Row 3 - column types.

    const std::vector<std::string>& types = data.colTypes();
    for (size_t c=0; c<ncols; ++c)
      log << "Column " << c << ", type = " << types[c] << "\n";

    // Data rows.

    for(size_t row=0; row<nrows; ++row) {
      log << "\nRow " << row << "\n";
      DBDataset::DBRow dbrow = data.getRow(row);

      // Loop over columns.

      for(size_t col=0; col<ncols; ++col) {
        if(types[col] == "bigint" || types[col] == "integer" || types[col] == "boolean") {
          long value = dbrow.getLongData(col);
          log << names[col] << " = " << value << "\n";
        }
        else if(types[col] == "real") {
          double value = dbrow.getDoubleData(col);
          log << names[col] << " = " << value << "\n";
        }
        else if(types[col] == "text" or types[col] == "boolean") {
          std::string value = dbrow.getStringData(col);
          log << names[col] << " = " << value << "\n";
        }
        else {
          mf::LogError("DBFolder") << "Unknown type " << types[col] << "\n";
          throw cet::exception("DBFolder") << "Unknown type.";
        }
      }
    }
  }

  bool DBFolder::CompareDataset(const DBDataset& data1, const DBDataset& data2) const
  {
    bool compare_ok = true;
    mf::LogInfo("DBFolder") << "\nComparing datasets." << "\n";

    size_t nrows1 = data1.nrows();
    size_t nrows2 = data2.nrows();
    //mf::LogInfo log("DBFolder");
    //log << "Dataset 1 contains " << nrows1 << " rows." << "\n";
    //log << "Dataset 2 contains " << nrows2 << " rows." << "\n";
    if(nrows1 != nrows2)
      compare_ok = false;

    // Compare begin time.

    std::string begin1 = data1.beginTime().DBStamp();
    std::string begin2 = data2.beginTime().DBStamp();
    if(begin1 != begin2)
      compare_ok = false;

    // Compare end time.

    std::string end1 = data1.beginTime().DBStamp();
    std::string end2 = data2.beginTime().DBStamp();
    if(end1 != end2)
      compare_ok = false;

    // Compare column names.

    size_t ncols1 = data1.ncols();
    size_t ncols2 = data2.ncols();
    const std::vector<std::string>& names1 = data1.colNames();
    const std::vector<std::string>& names2 = data2.colNames();
    if(ncols1 != ncols2 || ncols1 != names1.size() || ncols2 != names2.size()) {
      mf::LogWarning("DBFolder") << "Columns names size mismatch " << ncols1
                                 << " vs. " << ncols2
                                 << " vs. " << names1.size()
                                 << " vs. " << names2.size()
                                 << "\n";
      compare_ok = false;
    }
    if(compare_ok) {
      for (size_t c=0; c<ncols1; ++c) {
        if(names1[c] != names2[c]) {
          mf::LogWarning("DBFolder") << "Name mismatch " << names1[c] << " vs. " << names2[c] << "\n";
          compare_ok = false;
        }
      }
    }

    // Compare column types.

    const std::vector<std::string>& types1 = data1.colTypes();
    const std::vector<std::string>& types2 = data2.colTypes();
    if(ncols1 != ncols2 || ncols1 != types1.size() || ncols2 != types2.size()) {
      mf::LogWarning("DBFolder") << "Column types ize mismatch " << ncols1
                                 << " vs. " << ncols2
                                 << " vs. " << types1.size()
                                 << " vs. " << types2.size()
                                 << "\n";
      compare_ok = false;
    }
    if(compare_ok) {
      for (size_t c=0; c<ncols2; ++c) {

        // Type "bigint" matches "integer."
        // Type "boolean" matches "integer."

        std::string type1 = types1[c];
        std::string type2 = types2[c];
        if(type1 == "bigint" || type1 == "boolean")
          type1 = "integer";
        if(type2 == "bigint" || type2 == "boolean")
          type2 = "integer";
        if(type1 != type2) {
          mf::LogWarning("DBFolder") << "Type mismatch " << type1 << " vs. " << type2 << "\n";
          compare_ok = false;
        }
      }
    }

    // Compare channels.

    const std::vector<DBChannelID_t>& channels1 = data1.channels();
    const std::vector<DBChannelID_t>& channels2 = data2.channels();
    if(nrows1 != nrows2 || nrows1 != channels1.size() || nrows2 != channels2.size()) {
      mf::LogWarning("DBFolder") << "Channels size mismatch " << nrows1
                                 << " vs. " << nrows2
                                 << " vs. " << channels1.size()
                                 << " vs. " << channels2.size()
                                 << "\n";
      compare_ok = false;
    }
    if(compare_ok) {
      for (size_t r=0; r<nrows1; ++r) {
        if(channels1[r] != channels2[r]) {
          mf::LogWarning("DBFolder") << "Channel mismatch " << channels1[r] << " vs. " << channels2[r] << "\n";
          compare_ok = false;
        }
      }
    }

    // Compare number of values.

    if(data1.data().size() != data2.data().size()) {
      mf::LogWarning("DBFolder") << "Values size mismatch " << data1.data().size()
                                 << " vs. " << data2.data().size()
                                 << "\n";
      compare_ok = false;
    }

    // Data rows.
    if(compare_ok) {
      for(size_t row=0; row<nrows1; ++row) {

        DBDataset::DBRow dbrow1 = data1.getRow(row);
        DBDataset::DBRow dbrow2 = data2.getRow(row);
        //mf::LogInfo("DBFolder") << "\nRow " << row << "\n";

        // Loop over columns.

        for(size_t col=0; col<ncols1; ++col) {
          if(types1[col] == "integer" || types1[col] == "bigint" ||  types1[col] == "boolean") {
            long value1 = dbrow1.getLongData(col);
            long value2 = dbrow2.getLongData(col);
            //mf::LogInfo log("DBFolder")
            //log << names1[col] << " 1 = " << value1 << "\n";
            //log << names2[col] << " 2 = " << value2 << "\n";
            if(value1 != value2) {
              mf::LogWarning("DBFolder") << "Value mismatch " << value1 << " vs. " << value2 << "\n";
              compare_ok = false;
            }
          }
          else if(types1[col] == "real") {
            double value1 = dbrow1.getDoubleData( col);
            double value2 = dbrow2.getDoubleData( col);
            //mf::LogInfo log("DBFolder")
            //log << names1[col] << " 1 = " << value1 << "\n";
            //log << names2[col] << " 2 = " << value2 << "\n";
            if(value1 != value2) {
              mf::LogWarning("DBFolder") << "Value mismatch " << value1 << " vs. " << value2 << "\n";
              compare_ok = false;
            }
          }
          else if(types1[col] == "text") {
            std::string value1 = dbrow2.getStringData(col);
            std::string value2 = dbrow2.getStringData(col);
            if(value1 != value2) {
              mf::LogWarning("DBFolder") << "Value mismatch " << value1 << " vs. " << value2 << "\n";
              compare_ok = false;
            }
          }
          else {
            mf::LogError("DBFolder") << "Unknown type " << types1[col] << "\n";
            throw cet::exception("DBFolder") << "Unknown type.";
          }
        }
      }
    }

    if(compare_ok) {
      mf::LogInfo("DBFolder") << "Comparison OK.\n" << "\n";
    }
    else{
      mf::LogError("DBFolder") << "Comparison fail." << "\n";
      throw cet::exception("DBFolder") << "Comparison fail.";
    }
    return compare_ok;
  }

}//end namespace lariov