SQLiteDB.cpp

//Created by KVClassFactory on Mon Jul 25 15:33:56 2016
//Author: John Frankland,,,
 
#include "SQLiteDB.h"
#include <TList.h>
#include "TSQLRow.h"
#include "TSQLTableInfo.h"
#include "TSQLColumnInfo.h"
#include <KVString.h>
#include <iostream>
#include "KVError.h"
#include <iomanip>
#include "TSystem.h"
#include <thread>
#include <chrono>
using namespace std::chrono_literals;
 
ClassImp(KVSQLite::database)
ClassImp(KVSQLite::table)
ClassImp(KVSQLite::column)
 
 
namespace KVSQLite {
 
   bool database::debug = false;
 
   // static maps instantiation
   std::map<TString, KVSQLite::column_type::types> table::type_map;
   std::map<KVSQLite::column_type::types, TString> column::inv_type_map;
 
 
 
   std::unique_ptr<TSQLResult> database::SelectRowsFromTable(const TString& table, const TString& columns, const TString& condition) const
   {
      // \param[in] table name of table
      // \param[in] columns comma-separated list of columns
      // \param[in] condition selection to be applied, if any
      // \returns result of query `SELECT [columns] FROM [table] WHERE [condition]`
 
      TString _query = "SELECT " +  double_quote_all_tables_and_columns(columns) + " FROM " + double_quote_all_tables_and_columns(table);
      if (condition != "") _query += " WHERE " + double_quote_all_tables_and_columns(condition);
      return query(_query);
   }
 
 
 
 
   void database::read_table_infos()
   {
      // initialise map of database tables from existing database
      std::unique_ptr<TList> tl(fDBserv->GetTablesList());
      TObject* o;
      TIter it_tab(tl.get());
      while ((o = it_tab())) {
 
         table t(o->GetName());
 
         std::unique_ptr<TSQLResult> columnRes(fDBserv->GetColumns("", o->GetName()));
         if (!columnRes.get()) {
            Error("read_table_infos", "Cannot get information on columns for table %s", o->GetName());
            return;
         }
         std::unique_ptr<TSQLRow> columnRow(columnRes->Next());
         while ((columnRow.get()) != nullptr) {
            column& col = t.add_column(columnRow->GetField(1), columnRow->GetField(2));
            TString primary_key(columnRow->GetField(5));
            if (primary_key.Atoi() == 1) t.primary_key(col.name());
            columnRow.reset(columnRes->Next());
         }
         fTables.insert(std::pair<std::string, KVSQLite::table>(o->GetName(), t));
         (*this)[t.name()].fDB = this;// set pointer to dB in table
      }
      // retrieve foreign key constraints for tables
      for (auto& p : fTables) {
         TString qry = Form("PRAGMA foreign_key_list('%s')", p.second.name());
         // produces output like:
         // id | seq | parent_table | child_key | parent_key | on_update | on_delete | match
         auto result = query(qry);
         std::unique_ptr<TSQLRow> row(result->Next());
         while (row) {
            TString parent_table(row->GetField(2));
            TString child_key(row->GetField(3));
            TString parent_key(row->GetField(4));
            p.second.foreign_key(child_key, parent_table, parent_key);
            row.reset(result->Next());
         }
         p.second.set_foreign_key_back_references();
      }
   }
 
 
 
 
   void database::show_tables() const
   {
      // print list of tables
      std::cout << "Tables in database:" << std::endl;
      for (auto it = fTables.begin();
            it != fTables.end(); ++it) {
         std::cout << "\t" << it->first << std::endl;
      }
   }
 
 
 
 
   void database::open(const TString& dbfile)
   {
      // Open/create sqlite db file given path
      //
      // Any special characters/environment variables are first expanded, so
      // you can use:
      //~~~~
      // ~/mydata.db
      // $(SOME_PATH)/toto.db
      //~~~~
 
      TString exp_path = dbfile;
      if (gSystem->ExpandPathName(exp_path)) {
         Error("open", "problem with SQLite database filename: %s", dbfile.Data());
         fIsValid = false;
         return;
      }
      TString uri = "sqlite://" + exp_path;
      fDBserv.reset(new KVSQLiteServer(uri, 0, 0));
      // check for valid database file
      if (!execute("pragma schema_version")) {
         if (fDBserv->IsLocked()) {
            // the db file has been locked by another process/thread
            // sleep, then try again (5 times)
            int i = 1;
            do {
               Info("open", "Database is locked. Try again in 1s...");
               std::this_thread::sleep_for(1s);
               Info("open", "Try number %d...", i);
               ++i;
            }
            while (!execute("pragma schema_version") && fDBserv->IsLocked() && (i < 10));
            if ((i < 10) && fDBserv->IsOK()) {
               Info("open", "...database successfully opened");
               fIsValid = true;
               read_table_infos();
               return;
            }
         }
         fDBserv->Close();
         fIsValid = false;
         return;
      }
      fIsValid = true;
      read_table_infos();
   }
 
 
 
 
   void database::PrintResults(TSQLResult* tabent, int column_width) const
   {
      int nfields = tabent->GetFieldCount();
      std::unique_ptr<TSQLRow> row(nullptr);
      for (int r = -1; true; ++r) {
         if (r > -1) {
            row.reset(tabent->Next());
            if (row.get() == nullptr) break;
            std::cout << std::setw(6) << r;
         }
         for (int f = 0; f < nfields; ++f) {
            if (r < 0) {
               if (f == 0) std::cout << std::setw(6) << "#";
               std::cout << "|" << std::setw(column_width) << tabent->GetFieldName(f) ;
            }
            else {
               std::cout << "|" << std::setw(column_width) << row->GetField(f) ;
            }
         }
         std::cout << "\n";
      }
   }
 
 
 
 
   void database::Dump() const
   {
      // Print on stdout contents of database
 
      std::cout << "Database : " << fDBserv->GetDB() << " [" << fDBserv->GetDBMS() << "]\n";
      std::unique_ptr<TList> tl(fDBserv->GetTablesList());
      TObject* o;
      TIter it_tab(tl.get());
      while ((o = it_tab())) {
         std::cout << "\t";
         std::cout << "Table : " << o->GetName() << "\n";
         std::unique_ptr<TSQLResult> tabent = SelectRowsFromTable(o->GetName());
         PrintResults(tabent.get());
      }
      std::cout << std::endl;
   }
 
 
 
   void database::print_selection(const TString& table, const TString& columns, const TString& condition, int column_width) const
   {
      // Print on stdout contents of database
 
      std::unique_ptr<TSQLResult> tabent = SelectRowsFromTable(table, columns, condition);
      PrintResults(tabent.get(), column_width);
      std::cout << std::endl;
   }
 
 
 
 
   table& database::add_table(const table& t)
   {
      // add table to database (if it does not exist already)
      //
      // returns reference to new table in database
      //
      // _WARNING:_ after calling this method, do not use the table given as argument
      //
      //     it does not correspond to the table in the database
      //
      //     instead use `db["table name"]` or the reference returned by this method to access the table
      //
      //~~~~{.cpp}
      //   e.g. KVSQLite::table tt("some table");
      //        tt.add_column(...);
      //        db.add_table(tt);
      //        db["some table"]["column"].set_data(...);
      //~~~~
 
      auto command = t.get_table_creation_command();
      if (debug)
         std::cout << command << std::endl;
      if (execute(command)) {
         fTables.insert(std::pair<std::string, KVSQLite::table>(t.name(), t));
         (*this)[t.name()].fDB = this;// set pointer to dB in table
         (*this)[t.name()].set_foreign_key_back_references();
      }
      return (*this)[t.name()];
   }
 
 
 
 
   bool database::prepare_data_insertion(const TString& table)
   {
      // Call this method before insert_data_row() in order to perform bulk data
      // insertion operation. Note that this can only be used in cases where values are set for
      // each column (apart from, eventually, a rowid alias column i.e. an INTEGER PRIMARY KEY)
      // before insertion. If not, use prepare_data_insertion_selected_columns().
      //
      //~~~~{.cpp}
      //   db.prepare_data_insertion("my_table");
      //   while(...){  // loop over data to insert
      //       // set up data in table
      //       db.insert_data_row();
      //   }
      //   db.end_data_insertion();  // terminate data insertion
      //~~~~
      // Until method end_data_insertion() is called, you cannot call prepare_data_insertion()
      // with a different table name.
 
      if (debug) Info("prepare_data_insertion", "table=%s", table.Data());
      if (fInserting) {
         if (fBulkTable) {
            Error("database::prepare_data_insertion",
                  "bulk insertion in progress for table %s; call database::commit() to terminate transaction",
                  fBulkTable->name());
            return false;
         }
         else {
            Error("database::prepare_data_insertion",
                  "bulk insertion in progress; call database::commit() to terminate transaction");
            return false;
         }
      }
      if (fSelecting) {
         Error("database::prepare_data_insertion",
               "data retrieval in progress; call get_next_result() until it returns false");
         return false;
      }
      fInserting = true;
      fDBserv->StartTransaction();
      // set up SQL statement for data insertion into table
      fBulkTable = &fTables[table.Data()];
      TString com(fBulkTable->get_insert_command());
      int ncol = fBulkTable->number_of_columns();
      int idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (idx) com += ",";
         if (!(*fBulkTable)[i].rowid_alias()) {
            com += Form("\"%s\"", (*fBulkTable)[i].name());;
            ++idx;
         }
      }
      com += ") VALUES (";
      idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (idx) com += ",";
         if (!(*fBulkTable)[i].rowid_alias()) {
            com += "?";
            ++idx;
         }
      }
      com += ")";
      if (debug) std::cout << com << std::endl;
      return prepare_statement(com);
   }
 
 
 
 
   bool database::prepare_data_insertion_selected_columns(const TString& table)
   {
      // Call this method before insert_data_row_selected_columns() in order to perform bulk data
      // insertion operation without giving values for all columns (apart from an eventual rowid alias column).
      //
      //~~~~{.cpp}
      //   // initialise columns
      //   db["my_table"].prepare_insert_single_row();
      //   // set data in columns for first row to insert
      //   db["my_table"]["colA"] = "ok";
      //   db["my_table"]["colC"] = 116;
      //   // set up bulk insertion
      //   db.prepare_data_insertion_selected_columns("my_table");
      //   // insert first row of data
      //   db.insert_data_row();
      //   while(...){  // loop over rest of data to insert
      //       // set up data in table
      //       db["my_table"]["colA"] = "bad";
      //       db["my_table"]["colC"] = -356;
      //       db.insert_data_row();
      //   }
      //   db.end_data_insertion();  // terminate data insertion
      //~~~~
      // Until method end_data_insertion() is called, you cannot call prepare_data_insertion()
      // with a different table name.
 
      if (debug) Info("prepare_data_insertion_selected_columns", "table=%s", table.Data());
      if (fInserting) {
         if (fBulkTable) {
            Error("database::prepare_data_insertion",
                  "bulk insertion in progress for table %s; call database::commit() to terminate transaction",
                  fBulkTable->name());
            return false;
         }
         else {
            Error("database::prepare_data_insertion",
                  "bulk insertion in progress; call database::commit() to terminate transaction");
            return false;
         }
      }
      if (fSelecting) {
         Error("database::prepare_data_insertion",
               "data retrieval in progress; call get_next_result() until it returns false");
         return false;
      }
      fInserting = true;
      fDBserv->StartTransaction();
      // set up SQL statement for data insertion into table
      fBulkTable = &fTables[table.Data()];
      TString com(fBulkTable->get_insert_command());
      int ncol = fBulkTable->number_of_columns();
      int idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (idx) com += ",";
         if (!(*fBulkTable)[i].rowid_alias() && (*fBulkTable)[i].has_data) {
            com += Form("\"%s\"", (*fBulkTable)[i].name());;
            ++idx;
         }
      }
      com += ") VALUES (";
      idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (idx) com += ",";
         if (!(*fBulkTable)[i].rowid_alias() && (*fBulkTable)[i].has_data) {
            com += "?";
            ++idx;
         }
      }
      com += ")";
      if (debug) std::cout << com << std::endl;
      return prepare_statement(com);
   }
 
 
 
 
   const char* table::get_insert_command() const
   {
      switch (fInsert) {
         case KVSQLite::insert_mode::FAIL:
            return Form("INSERT OR FAIL INTO \"%s\"(", name());
            break;
         case KVSQLite::insert_mode::IGNORE:
            return Form("INSERT OR IGNORE INTO \"%s\"(", name());
            break;
         case KVSQLite::insert_mode::REPLACE:
            return Form("INSERT OR REPLACE INTO \"%s\"(", name());
            break;
         default:
         case KVSQLite::insert_mode::DEFAULT:
            break;
      }
      return Form("INSERT INTO \"%s\"(", name());
   }
 
 
 
 
   TString column::get_declaration() const
   {
      // return declaration for column, including type & constraint
      //
      // the name of the column is enclosed in double-quotes, in case it is the
      // same as an SQL keyword
 
      TString decl;
      decl.Form("\"%s\" %s", name(), TypeName().Data());
      if (!fConstraint.IsNull()) {
         decl += " ";
         decl += fConstraint;
      }
      return decl;
   }
 
 
 
 
   void database::insert_data_row()
   {
      // Call (repeatedly) after a call to prepare_data_insertion()
      // in order to insert current contents of table columns as a new row in the database.
      //
      // Value of each column should first be set like this:
      //
      //~~~~{.cpp}
      //    db["table"]["id"].set_data(6);
      //    db["table"]["name"] = "triumph";
      //~~~~
      // Call end_data_insertion() when all data has been inserted
 
      if (!fInserting || !fBulkTable) {
         Error("database::insert_data_row",
               "no transaction initialized; call prepare_data_insertion(name_table) before this method");
         return;
      }
      if (fSelecting) {
         Error("database::insert_data_row",
               "data retrieval in progress; call get_next_result() until it returns false, then call prepare_data_insertion(name_table) before this method");
         return;
      }
      int ncol = fBulkTable->number_of_columns();
      fSQLstmt->NextIteration();
      int idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (!(*fBulkTable)[i].rowid_alias()) {
            (*fBulkTable)[i].set_data_in_statement(fSQLstmt.get(), idx);
            ++idx;
         }
      }
   }
 
 
 
   void database::insert_data_row_selected_columns()
   {
      // Call (repeatedly) after a call to prepare_data_insertion()
      // in order to insert current contents of table columns as a new row in the database.
      //
      // Value of each column should first be set like this:
      //
      //~~~~{.cpp}
      //    db["table"]["id"].set_data(6);
      //    db["table"]["name"] = "triumph";
      //~~~~
      // Call end_data_insertion() when all data has been inserted
 
      if (!fInserting || !fBulkTable) {
         Error("database::insert_data_row",
               "no transaction initialized; call prepare_data_insertion(name_table) before this method");
         return;
      }
      if (fSelecting) {
         Error("database::insert_data_row",
               "data retrieval in progress; call get_next_result() until it returns false, then call prepare_data_insertion(name_table) before this method");
         return;
      }
      int ncol = fBulkTable->number_of_columns();
      fSQLstmt->NextIteration();
      int idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (!(*fBulkTable)[i].rowid_alias() && (*fBulkTable)[i].has_data) {
            (*fBulkTable)[i].set_data_in_statement(fSQLstmt.get(), idx);
            ++idx;
         }
      }
   }
 
 
 
 
   void database::end_data_insertion()
   {
      // Call after prepare_data_insertion() & insert_data_row() have been
      // used to insert data into a table
 
      if (debug) Info("end_data_insertion", "now");
      if (!fInserting) {
         Error("database::end_data_insertion",
               "no transaction initialized; call prepare_data_insertion(name_table) first");
         return;
      }
      if (fSelecting) {
         Error("database::insert_data_row",
               "data retrieval in progress; call get_next_result() until it returns false, then call prepare_data_insertion(name_table) before this method");
         return;
      }
      fBulkTable = nullptr;
      fSQLstmt->Process();
      fDBserv->Commit();
      fInserting = false;
   }
 
 
 
   void database::print_selected_data(const TString& tables, const TString& columns, const TString& selection, bool distinct, const TString& anything_else)
   {
      // Print out results of a call to select_data().
 
      if (select_data(tables, columns, selection, distinct, anything_else)) {
         // retrieve column titles (with table name) and store length of each
         std::vector<TString> col_titles;
         std::vector<int> col_width;
         std::for_each(std::begin(fSQLstmtCols), std::end(fSQLstmtCols),
                       [&col_titles, &col_width]
         (const column * col) {
            col_titles.push_back(col->get_table() + "::" + col->Name());
            col_width.push_back(col->get_table().Length() + 2 + col->Name().Length());
         }
                      );
         // retrieve column data and update maximum length of column
         using data_row = std::vector<TString>;
         std::vector<data_row> col_data;
         while (get_next_result()) {
            data_row current_data_row;
            int col_index = 0;
            std::for_each(std::begin(fSQLstmtCols), std::end(fSQLstmtCols),
                          [&col_index, &col_width, &current_data_row]
            (const column * col) {
               current_data_row.push_back(col->data().GetTString());
               col_width[col_index] = std::max(current_data_row[col_index].Length(), col_width[col_index]);
               ++col_index;
            }
                         );
            col_data.push_back(current_data_row);
         }
         auto print_line = [&col_width]() {
            for (auto width : col_width) {
               std::cout << "+-";
               for (int i = 0; i <= width; ++i) std::cout << "-";
            }
            std::cout << "+\n";
         };
         auto print_column_item = [](const TString & item, int width, bool end_line = false) {
            std::cout << "| " << item;
            for (int i = 0; i <= width - item.Length(); ++i)
               std::cout << " ";
            if (end_line)
               std::cout << "|";
         };
         auto print_column_data = [&col_width, &print_column_item](const std::vector<TString>& cols) {
            int col_index = 0;
            int ncols = cols.size();
            for (auto& col : cols) {
               print_column_item(col, col_width[col_index], (col_index == ncols - 1));
               ++col_index;
            }
            std::cout << std::endl;
         };
         print_line();
         print_column_data(col_titles);
         print_line();
         for (auto& row : col_data) print_column_data(row);
         print_line();
      }
   }
 
 
 
 
   std::list<const table*> database::generate_table_selection_string(const TString& tables, KVString& table_selection) const
   {
      std::list<const table*> table_list;
      KVString _tables(tables);
      _tables.Begin(",");
      // use double-quoted table names in table_selection
      int i(0);
      std::unordered_map<std::string, int> table_names;
      while (!_tables.End()) {
         TString tabnam = _tables.Next();
         table_list.push_back(& operator[](tabnam));
         if (i) {
            table_selection += " INNER JOIN \"" + tabnam + "\"";
            // add any foreign key constraints involving any previously mentioned table
            auto this_table = table_list.back();
            if (this_table->has_foreign_keys(false)) { // include back references
               TString on_and = "ON";
               for (auto& f : this_table->fForeignKeys) {
                  // make sure other table is already in list
                  if (table_names[f.parent_table.Data()]) {
                     table_selection += Form(" %s %s.%s = %s.%s",
                                             on_and.Data(),
                                             this_table->name(), f.child_key.Data(),
                                             f.parent_table.Data(), f.parent_key.Data());
                     on_and = "AND"; // chain successive foreign key constraints: 'ON a=b AND c=d AND ...'
                  }
               }
            }
         }
         else
            table_selection += "\"" + tabnam + "\"";
         ++table_names[tabnam.Data()];
         ++i;
      }
      return table_list;
   }
 
 
 
 
   KVString KVSQLite::database::double_quote_all_tables_and_columns(const TString& input) const
   {
      auto replacer = [&](const KVString & repl) {
         KVString replaced;
         // treat each whitespace-separated token in string separately
         repl.Begin(" ");
         while (!repl.End()) {
            auto tok = repl.Next();
            if (replaced.Length()) replaced += " ";
            bool stop_replacing = false;
            for (auto& tp : fTables) {
               TString table_name = tp.second.name();
               if (tok == table_name) {
                  tok.Prepend("\"");
                  tok.Append("\"");
                  stop_replacing = true;
                  break;
               }
               for (auto& c : tp.second.fColumns) {
                  TString column_name = c.name();
                  if (tok == column_name) {
                     tok.Prepend("\"");
                     tok.Append("\"");
                     stop_replacing = true;
                     break;
                  }
               }
               if (stop_replacing) break;
            }
            replaced += tok;
         }
         return replaced;
      };
      // no replacements inside string constants i.e. between pairs of ''
      KVString _input(input);
      _input.Begin("'");
      bool in_quotes = _input.BeginsWith("'");
      KVString output;
      while (!_input.End()) {
         auto part = _input.Next();
         if (!in_quotes) {
            output += replacer(part);
         }
         else {
            output += "'" + part + "'";
         }
         in_quotes = !in_quotes;
      }
      return output;
   }
 
 
 
 
   bool database::select_data(const TString& tables, const TString& columns, const TString& selection, bool distinct, const TString& anything_else) const
   {
      // Select data in database from given table(s) according to
      //~~~~
      //    SELECT [columns] FROM [tables] WHERE [selection] [anything_else]
      //~~~~
      // In order to retrieve results, call get_next_result() until it returns false.
      //
      // \param tables if more than 1 table is given, separate table names with commas.
      //           if 1 table has a foreign key referencing the other, this allows to JOIN data
      //           in both tables together. [columns] can then refer to columns in either table.
      // \param columns ="*" by default, i.e. data from all columns is retrieved.
      //            If specific column data is to be selected, give a comma-separated list of
      //            column names. These will be quoted correctly in case they contain spaces.
      // \param distinct can be used in conjunction with a selection of specific columns in order
      //             to retrieve only rows of data with different values for the column(s).
      //
      // You can use the COUNT() function as one of the column names (upper or lower case).
      // In this case, the value for each row can be retrieved by calling get_count_column_value()
      // after each call to get_next_value().
 
      if (debug) Info("select_data", "tables=%s columns=%s selection=%s anything_else=%s",
                         tables.Data(), columns.Data(), selection.Data(), anything_else.Data());
      if (fInserting) {
         Error("database::select_data",
               "data insertion in progress; call end_data_insertion() before retrieving data");
         return false;
      }
      if (fSelecting) {
         Error("database::select_data",
               "data retrieval already in progress; call get_next_result() until it returns false before making new selection");
         return false;
      }
 
      fSQLstmtCols.clear();
 
      KVString table_selection;
      std::list<const table*> table_list = generate_table_selection_string(tables, table_selection);
 
      KVString column_selection(""), _columns(columns);
      if (columns == "*") {
         column_selection = "*";
         distinct = false; // don't allow 'SELECT DISTINCT * FROM ....' (?)
         // find right column in right table for each item
         // loop over all columns in each table
         for (auto tabs : table_list) {
            for (int i = 0; i < tabs->number_of_columns(); ++i) {
               fSQLstmtCols.push_back(&(*tabs)[i]);
            }
         }
      }
      else {
         if (distinct) column_selection = "DISTINCT ";
         _columns.Begin(",");
         int i(0);
         while (!_columns.End()) {
            TString colnam = _columns.Next();
            if (i) column_selection += ", ";
            ++i;
            // find right column in right table for this item
            TString COLNAM = colnam;
            COLNAM.ToUpper();
            if (COLNAM.BeginsWith("COUNT(")) { // look for use of COUNT()
               // no double quotes around COUNT()!
               column_selection += Form("%s", colnam.Data());
               fSQLstmtCols.push_back(&count_column);
            }
            else {
               // put double-quoted column names in column_selection
               column_selection += Form("\"%s\"", colnam.Data());
               for (auto tabs : table_list)
                  if (tabs->has_column(colnam)) fSQLstmtCols.push_back(&(*tabs)[colnam]);
            }
         }
      }
 
      TString cond = Form("SELECT %s FROM %s", column_selection.Data(), table_selection.Data());
      if (selection != "") {
         // make sure any column names or table names are double quoted
         cond += " WHERE " + double_quote_all_tables_and_columns(selection);
      }
      if (anything_else != "") {
         // make sure any column names or table names are double quoted
         cond += " " + double_quote_all_tables_and_columns(anything_else);
      }
      if (debug)
         std::cout << cond << std::endl;
      prepare_statement(cond);
      if (fSQLstmt.get() == nullptr) {
         Error("database::select_data", "problem processing : %s", cond.Data());
         fSelecting = false;
         fBulkTable = nullptr;
         return false;
      }
      fSQLstmt->EnableErrorOutput();
      if (fSQLstmt->Process()) {
         fSQLstmt->StoreResult();
         fSelecting = true;
         fEmptyResultSet = false;
         return true;
      }
      else if (!fSQLstmt->IsError()) {
         // query ok, no results correspond to selection
         fSQLstmt->StoreResult();
         fSelecting = true;
         fEmptyResultSet = true;
         return true;
      }
      fSelecting = false;
      fBulkTable = nullptr;
      return false;
   }
 
 
 
 
   bool database::select_data_in_list(const TString& tables, const KVNameValueList& selection, const TString& columns, bool distinct, const TString& anything_else) const
   {
      // Select data according to the name-value pairs in the KVNameValueList. Each name in the list is assumed to correspond to a column name,
      // with a corresponding value of the correct type.
      //
      // For other arguments, see the other select_data() method.
 
      return select_data(tables, columns, selection.AsSQLSelection(), distinct, anything_else);
   }
 
 
 
 
   bool database::get_next_result() const
   {
      // Retrieve next result row resulting from previous call to select_data()
      // \returns kFALSE when no more data is retrieved
 
      if (debug) Info("get_next_result", "now");
      if (fInserting) {
         Error("database::get_next_result",
               "data insertion in progress; call end_data_insertion() then select_data() before this method");
         return false;
      }
      if (!fSelecting) {
         Error("database::get_next_result",
               "no data retrieval in progress; select_data() must be called and return true before calling this method");
         return false;
      }
      if (!fEmptyResultSet && fSQLstmt->NextResultRow()) {
         // set column data
         int idx = 0;
         for (auto col : fSQLstmtCols) {
            col->set_data_from_statement(fSQLstmt.get(), idx);
            ++idx;
         }
         return kTRUE;
      }
      fBulkTable = nullptr;
      fSelecting = false;
      return kFALSE;
   }
 
 
 
 
   KVNumberList database::get_integer_list(const TString& tables, const TString& column, const TString& selection, const TString& anything_else)
   {
      // \note Only for INTEGER columns!
      // Fill KVNumberList with all `DISTINCT` values of "column" (only 1 column name at a time) for given selection
 
      if (column.Contains(",")) {
         Error("database::get_integer_list", "Only 1 column name at a time please!");
         return {};
      }
      KVNumberList result;
      if (select_data(tables, column, selection, true, anything_else)) {
         auto column_data = fSQLstmtCols.front();
         while (get_next_result()) {
            result.Add(column_data->get_data<int>());
         }
      }
      return result;
   }
 
 
 
 
   TString database::get_string_list(const TString& tables, const TString& column, const TString& selection, const TString& anything_else)
   {
      // \note Only for TEXT columns!
      // Fill TString with comma-separated list of values of "column" (only 1 column name at a time) for given selection
      // \note Any NULL entries will be ignored
 
      if (column.Contains(",")) {
         Error("database::get_string_list", "Only 1 column name at a time please!");
         return "";
      }
      TString result;
      if (select_data(tables, column, selection, false, anything_else)) {
         auto column_data = fSQLstmtCols.front();
         while (get_next_result()) {
            if (column_data->is_null()) continue;
            if (result != "") result += ",";
            result += column_data->get_data<TString>();
         }
      }
      return result;
   }
 
 
 
 
   KVNameValueList database::get_name_value_list(const TString& tablename, const TString& name_column, const TString& value_column, const TString& selection, const TString& anything_else)
   {
      // Fill KVNameValueList with selected rows from table, adding for each row a parameter with the
      // name contained in "name_column" (must be of type `TEXT`) and the value contained in "value_column"
      // (can be `INTEGER`, `REAL`, or `TEXT`)
 
      KVNameValueList result;
      if (select_data(tablename, Form("%s,%s", name_column.Data(), value_column.Data()), selection, false, anything_else)) {
         auto it = fSQLstmtCols.begin();
         auto nom = *it++;
         auto val = *it;
         while (get_next_result()) {
            result.SetValue(nom->get_data<TString>(), val->data());
         }
      }
      return result;
   }
 
 
 
 
   void database::get_name_value_list_for_each_row(const TString& tablename, std::function<bool (const KVNameValueList&)> callback, const TString& selection, const TString& anything_else)
   {
      // For each row from the table corresponding to given selection (if any), we fill a KVNameValueList with all
      // non-null column values, then call the user's callback function
      //~~~{.cpp}
      // bool my_callback(const KVNameValueList&);
      //~~~
      //with it. If the callback returns false, we stop the iteration (but finish the SQL query).
 
      bool call_callback = true;
      KVNameValueList result;
      if (select_data(tablename, "*", selection, false, anything_else)) {
         while (get_next_result()) {
            for (auto col : fSQLstmtCols) {
               if (!col->is_null())
                  result.SetValue(col->name(), col->data());
            }
            if (call_callback)
               call_callback = callback(result);
         }
      }
   }
 
 
 
 
   TGraph* database::create_graph(const TString& tablename, const TString& Xcolumn, const TString& Ycolumn, const TString& selection)
   {
      // Create and fill a TGraph from values Xcolumn and Ycolumn in table,
      // using the selection if required
 
      if (select_data(tablename, Form("%s,%s", Xcolumn.Data(), Ycolumn.Data()), selection)) {
         TGraph* g = new TGraph;
         int i = 0;
         auto it = fSQLstmtCols.begin();
         auto Xcol = *it++;
         auto Ycol = *it;
         while (get_next_result()) {
            g->SetPoint(i++, Xcol->get_data<double>(), Ycol->get_data<double>());
         }
         return g;
      }
      return nullptr;
   }
 
 
 
 
   void database::clear_table(const TString& name)
   {
      // Delete all rows in table, but does not remove table
      delete_rows_in_table(name);
   }
 
 
 
 
   int database::count(const TString& tables, const TString& column, const TString& selection, bool distinct, const TString& anything_else) const
   {
      // Returns number of rows corresponding to equivalent 'select_data' call
      //
      //  + if column="*" all rows are included
      //  + if a column name is given, only rows with a non-NULL value for column are counted
      //  + if distinct=false, count all rows including those with the same value of column
      //  + if distinct=true, count the number of different values of column
      //
      // Only 1 column name can be given as second argument
      //
      // Examples
      //
      //~~~
      // count("some_table")
      //    => SELECT COUNT(*) FROM some_table;
      //
      // count("some_table", "some_column")
      //    => SELECT COUNT(some_column) FROM some_table;
      //
      // count("some_table", "some_column", "", true)
      //    => SELECT COUNT(DISTINCT some_column) FROM some_table;
      //~~~
 
      if (debug) Info("count", "tables=%s column=%s selection=%s anything_else=%s",
                         tables.Data(), column.Data(), selection.Data(), anything_else.Data());
      TString qry = "SELECT count(";
      if (distinct && column != "*") qry += "DISTINCT ";
      if (column != "*") qry += Form("\"%s\"", column.Data());
      else qry += "*";
      qry += ") FROM ";
      KVString table_selection;
      generate_table_selection_string(tables, table_selection);
      qry += table_selection;
      if (selection != "") {
         qry += " WHERE " + double_quote_all_tables_and_columns(selection);
      }
      if (anything_else != "") {
         qry += " " + double_quote_all_tables_and_columns(anything_else);
      }
      if (debug) std::cout << qry << std::endl;
      auto result = query(qry);
      std::unique_ptr<TSQLRow> row(result->Next());
      TString number = row->GetField(0);
      return number.Atoi();
   }
 
 
 
 
   bool database::update(const TString& table, const TString& columns, const TString& selection)
   {
      // update the given columns of an entry in the table corresponding to selection (if given)
      // the current values of the data members of the columns will be used
      //
      // This is equivalent to
      //~~~~
      //    UPDATE [table] SET col1=newval,col2=newval,... [WHERE [selection]]
      //~~~~
 
      if (debug) Info("update", "table=%s columns=%s selection=%s", table.Data(), columns.Data(), selection.Data());
      if (fInserting) {
         Error("database::update",
               "data insertion in progress; call end_data_insertion() before doing anything else");
         return false;
      }
      if (fSelecting) {
         Error("database::update",
               "data retrieval in progress; call get_next_result() until it returns false before doing anything else");
         return false;
      }
 
      fBulkTable = &fTables[table.Data()];
      TString query = Form("UPDATE \"%s\" SET ", table.Data());
      int ncol = fBulkTable->number_of_columns();
      int idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (columns.Contains((*fBulkTable)[i].name())) {
            if (idx) query += ",";
            query += Form("\"%s\"", (*fBulkTable)[i].name());
            query += "=?";
            ++idx;
         }
      }
      if (selection != "") query += " WHERE " + double_quote_all_tables_and_columns(selection);
      if (debug) std::cout << query << std::endl;
      if (!prepare_statement(query))
         return kFALSE;
      fSQLstmt->NextIteration();
      idx = 0;
      for (int i = 0; i < ncol; ++i) {
         if (columns.Contains((*fBulkTable)[i].name())) {
            (*fBulkTable)[i].set_data_in_statement(fSQLstmt.get(), idx);
            ++idx;
         }
      }
      return (fSQLstmt->Process());
   }
 
 
 
 
   void database::delete_table(const TString& table)
   {
      // Delete table from database. Equivalent to
      //
      //~~~~
      //   DROP TABLE IF EXISTS table
      //~~~~
      TString query = Form("DROP TABLE IF EXISTS %s", table.Data());
      execute(query);
   }
 
 
 
 
   void database::delete_rows_in_table(const TString& table, const TString& selection)
   {
      // Delete rows from the table corresponding to selection
      //
      // This is equivalent to
      //~~~~
      //    DELETE FROM [table] WHERE [selection]
      //~~~~
      // With no selection, deletes all rows of table (clear_table())
 
      if (debug) Info("delete_data", "table=%s selection=%s",
                         table.Data(), selection.Data());
      TString query = Form("DELETE FROM \"%s\"", table.Data());
      if (selection != "") query += " WHERE " + double_quote_all_tables_and_columns(selection);
      execute(query);
   }
 
 
 
 
   column& database::add_column(const TString& table, const TString& name, const TString& type)
   {
      // add column to existing table
      // \returns reference to new column
      TString query = Form("ALTER TABLE \"%s\" ADD COLUMN \"%s\" %s", table.Data(), name.Data(), type.Data());
      execute(query);
      return (*this)[table].add_column(name, type);
   }
 
 
 
 
   void database::add_missing_columns(const TString& _table_, const KVNameValueList& l)
   {
      // add to table any columns which are defined in the list but don't exist
      // \note cannot be called during data insertion or retrieval!!!
 
      if (fInserting) {
         Error("database::add_missing_columns",
               "data insertion in progress; call end_data_insertion() before doing anything else");
         return;
      }
      if (fSelecting) {
         Error("database::add_missing_columns",
               "data retrieval in progress; call get_next_result() until it returns false before doing anything else");
         return;
      }
 
      table& tab = (*this)[_table_];
      for (auto& par : l)
         if (!tab.has_column(par.GetName())) add_column(_table_, par.GetName(), par.GetSQLType());
   }
 
 
 
 
   void database::copy_table_data(const TString& source, const TString& destination, const TString& columns, const TString& selection)
   {
      // Copy all selected data in 'source' table to 'destination'
      //
      // If the columns of the two tables are not identical, specify the columns to copy in 'columns'
      // (comma-separated list)
      // \note SQLite will not allow copy if the number of selected columns from 'source' is not
      // exactly equal to the number of columns in 'destination'
 
      TString COLUMNS = columns;
      if (COLUMNS != "*") {
         // quote all column names
         COLUMNS = "";
         KVString _columns(columns);
         _columns.Begin(",");
         while (!_columns.End()) {
            if (COLUMNS != "") COLUMNS += ", ";
            COLUMNS += Form("\"%s\"", _columns.Next(kTRUE).Data());
         }
      }
      TString query = Form("INSERT INTO \"%s\" SELECT %s FROM \"%s\"", destination.Data(), COLUMNS.Data(), source.Data());
      if (selection != "") query += " WHERE " + double_quote_all_tables_and_columns(selection);
      execute(query);
   }
 
 
 
 
   void column::init_type_map()
   {
      inv_type_map[KVSQLite::column_type::REAL] = "REAL";
      inv_type_map[KVSQLite::column_type::INTEGER] = "INTEGER";
      inv_type_map[KVSQLite::column_type::TEXT] = "TEXT";
      inv_type_map[KVSQLite::column_type::TIMESTAMP] = "TEXT";
      inv_type_map[KVSQLite::column_type::BLOB] = "BLOB";
   }
 
 
 
 
   TString column::_type()
   {
      return inv_type_map[fNameType.second];
   }
 
 
 
 
   template<> void column::set_data(const KVNamedParameter& x)
   {
      fData.Set(x.GetName(), x);
      fIsNull = false;
      has_data = true;
   }
 
 
 
 
   void column::set_data_in_statement(TSQLStatement* s, int idx) const
   {
      // set value of parameter in SQLite statement corresponding to this column
      // \param idx if given, use it as the statement parameter index instead of
      //        the column's index in the table (case where not all columns are treated
      //        in the statement)
 
      if (idx < 0) idx = index();
      if (fIsNull) {
         // null parameter
         s->SetNull(idx);
         return;
      }
      switch (type()) {
         case KVSQLite::column_type::REAL:
            s->SetDouble(idx, fData.GetDouble());
            break;
         case KVSQLite::column_type::INTEGER:
            s->SetInt(idx, fData.GetInt());
            break;
         case KVSQLite::column_type::TEXT:
         case KVSQLite::column_type::TIMESTAMP:
            s->SetString(idx, fData.GetString(), -1);
            break;
         case KVSQLite::column_type::BLOB:
            s->SetBinary(idx, fBlob, fBlobSize);
            break;
         default:
            break;
      }
   }
 
 
 
   void column::set_data_from_statement(TSQLStatement* s, int idx) const
   {
      // set value of column according to value of parameter in statement
      //
      // any column which has a NULL value will be given value 0, 0.0 or ""
      // (for `INTEGER`, `REAL` or `TEXT` type, respectively): use column::is_null()
      // to check if this corresponds to a null column value.
      //
      // \param idx if given, use it as the statement parameter index instead of
      // the column's index in the table (case where not all columns are treated
      // in the statement)
 
      if (idx < 0) idx = index();
      fIsNull = s->IsNull(idx);
      switch (type()) {
         case KVSQLite::column_type::REAL:
            fData.Set(fIsNull ? 0.0 : s->GetDouble(idx));
            break;
         case KVSQLite::column_type::INTEGER:
            fData.Set(fIsNull ? 0 : s->GetInt(idx));
            break;
         case KVSQLite::column_type::TEXT:
         case KVSQLite::column_type::TIMESTAMP:
            fData.Set(fIsNull ? "" : s->GetString(idx));
            break;
         case KVSQLite::column_type::BLOB:
            if (fIsNull) {
               fBlobSize = 0;
            }
            else {
               if (!fBlob) fBlob = (void*) new unsigned char[256];
               s->GetBinary(idx, fBlob, fBlobSize);
            }
            break;
         default:
            break;
      }
   }
 
 
 
 
   TString column::get_data_string_for_insert()
   {
      // write data in string; for TEXT data we enclose in single quotes
      //
      // we also replace any "'" with "''" (otherwise SQL error on insert)
      if (fData.IsString()) {
         auto r = fData.GetTString();
         r.ReplaceAll("''", "~#%"); // in case there are already double apostrophes
         r.ReplaceAll("'", "''");
         r.ReplaceAll("~#%", "''"); // in case there are already double apostrophes
         return Form("'%s'", r.Data());
      }
      return fData.GetTString();
   }
 
 
 
 
   void table::init_type_map()
   {
      type_map["REAL"] = KVSQLite::column_type::REAL;
      type_map["INTEGER"] = KVSQLite::column_type::INTEGER;
      type_map["TEXT"] = KVSQLite::column_type::TEXT;
      type_map["TIMESTAMP"] = KVSQLite::column_type::TIMESTAMP;
      type_map["BLOB"] = KVSQLite::column_type::BLOB;
   }
 
 
 
 
   void table::set_foreign_key_back_references()
   {
      // called by database::add_table
      //
      // for any foreign key defined for this table, we set up a back reference in
      // the parent table
 
      for (auto& f : fForeignKeys)
         fDB->get_table(f.parent_table).fForeignKeys.push_back({f.parent_key, name(), f.child_key, true});
   }
 
 
 
 
   TString table::get_table_creation_command() const
   {
      // Create and fill SQLite command to create this table, i.e.
      //
      //~~~
      // CREATE TABLE name (
      //       col1 TYPE_1 [CONSTRAINT_1]
      //       col2 TYPE_2 [CONSTRAINT_2]
      //       ...
      //       [PRIMARY KEY(col1,...)]
      //       [FOREIGN KEY(col1,...) REFERENCES other_table(other_column) ...]
      // );
      //~~~
      TString command("CREATE ");
      if (is_temporary()) command += "TEMPORARY ";
      //command += "TABLE IF NOT EXISTS \"";
      command += "TABLE ";
      command += "\"" + fName + "\"";
      command += " (";
      for (int i = 0; i < number_of_columns(); ++i) {
         command += "\n   ";
         command += fColumns[i].get_declaration();
         if (i < number_of_columns() - 1) command += ",";
      }
      if (has_primary_key()) {
         command += ",\n   PRIMARY KEY(";
         command += "\"" + fPrimaryKey + "\"";
         command += ")";
      }
      if (has_foreign_keys()) {
         // foreign key constraints
         for (auto& fk : fForeignKeys) {
            command += ",\n   FOREIGN KEY(";
            command += "\"" + fk.child_key + "\"";
            command += ") REFERENCES ";
            command += "\"" + fk.parent_table + "\"";
            if (!fk.parent_key.IsNull()) {
               command += "(";
               command += "\"" + fk.parent_key + "\"";
               command += ")";
            }
         }
      }
      command += "\n);";
      return command;
   }
 
 
 
 
   void table::show_columns() const
   {
      // print list of columns
      std::cout << "Columns in table:" << std::endl;
#ifdef WITH_CPP11
      for (auto it = fColumns.begin(); it != fColumns.end(); ++it) {
#else
      for (std::vector<KVSQLite::column>::const_iterator it = fColumns.begin(); it != fColumns.end(); ++it) {
#endif
         std::cout << "\t" << it->name() << " [" << it->TypeName() << "]" << std::endl;
      }
   }
 
 
 
 
   column& table::add_column(const column& c)
   {
      // add column to table
      // \return reference to added column.
      // \note cannot be used for existing table in database: see database::add_column()
      fColumns.push_back(c);
      fColMap[c.name()] = c.index();
      fColumns.back().set_table(name());
      return fColumns.back();
   }
 
 
 
 
   column& table::add_column(const TString& name, const TString& type)
   {
      // add column to table
      // \return reference to added column
      // \note cannot be used for existing table in database: see database::add_column()
      return add_column(name, type_map[type]);
   }
 
 
 
 
   void table::foreign_key(const TString& child_key, const TString& parent_table, const TString& parent_key)
   {
      // add a FOREIGN KEY constraint to the table.
      //
      // The child key(s) should first be declared using add_column()
      //
      //~~~
      //  foreign_key("item_id", "items", "id")
      //
      //     => FOREIGN KEY(item_id) REFERENCES items(id)
      //
      //  foreign_key("item_id", "items")
      //
      //     => FOREIGN KEY(item_id) REFERENCES items
      //
      //         [case where 'items' has an INTEGER PRIMARY KEY, no need to specify]
      //
      //  foreign_key("songartist,songalbum", "album", "albumartist,albumname")
      //
      //     => FOREIGN KEY(songartist, songalbum) REFERENCES album(albumartist, albumname)
      //
      //         [case where both parent and child tables have composite primary keys]
      //~~~
 
      KVString check_columns = child_key;
      check_columns.Begin(",");
      bool ok = true;
      while (!check_columns.End()) {
         auto nxt = check_columns.Next(kTRUE);
         if (!has_column(nxt)) {
            Error("table::foreign_key", "Child key \"%s\" has not being declared with add_column()", nxt.Data());
            ok = false;
         }
      }
      if (!ok) return;
      fForeignKeys.push_back({child_key, parent_table, parent_key});
   }
 
 
 
 
   int table::check_columns(const KVNameValueList& l)
   {
      // make sure that all parameters in the list have corresponding columns in the table
      // \returns the number of columns to be added
 
      int ncols = 0;
      int ipar = l.GetNpar();
      for (int i = 0; i < ipar; ++i) {
         KVNamedParameter* par = l.GetParameter(i);
         if (!has_column(par->GetName())) ncols++;
      }
      return ncols;
   }
 
 
 
 
   void table::prepare_data(const KVNameValueList& l, const KVNamedParameter* null_value)
   {
      // fill all columns in table with data contained in KVNameValueList parameters having the same name.
      //
      // any columns which do not appear in the KVNameValueList (except for PRIMARY KEY) will be set to 'null'
      //
      // if required, any parameters with the same type&value as "null_value" will be set to 'null' too
 
      for (int i = 0; i < number_of_columns(); ++i) {
         KVNamedParameter* p = l.FindParameter((*this)[i].name());
         if (p && !(null_value && p->HasSameValueAs(*null_value)))
            (*this)[i].set_data(*p);
         else
            (*this)[i].set_null();
      }
   }
 
 
 
 
   void table::set_all_columns_null()
   {
      // set the value of all columns in the table to NULL
      for (int i = 0; i < number_of_columns(); ++i) {
         (*this)[i].set_null();
      }
   }
 
 
 
 
   TString table::get_column_names(const TString& exclude, const TString& delim) const
   {
      // Return a comma-separated list of the colum names
      //
      // \param[in] delim separator to use in list (default: ",")
      // \param[in] exclude list of column names to exclude from list
 
      TString namelist;
      int added = 0;
      for (int i = 0; i < number_of_columns(); ++i) {
         TString name = (*this)[i].name();
         if (exclude.Contains(name)) continue;
         if (added) namelist += delim;
         namelist += name;
         ++added;
      }
      return namelist;
   }
 
 
 
 
   void table::prepare_insert_single_row()
   {
      // call before setting data in individual columns and then calling insert_single_row()
      //
      //~~~{.cpp}
      //prepare_insert_single_row();
      //my_table["col1"] = 6;
      //my_table["col3"] = "hello";
      //insert_single_row();
      //~~~
 
      for (auto& col : fColumns) col.has_data = false;
   }
 
 
 
 
   void table::insert_single_row()
   {
      // use to add single row data to table. see prepare_insert_single_row() for usage.
 
      if (database::debug) Info("insert_single_row", "table=%s", name());
      TString query = get_insert_command();
      int ncols = 0;
      for (auto& col : fColumns) {
         if (col.has_data) {
            if (ncols) query += ",";
            query += "\"" + col.Name() + "\"";
            ++ncols;
         }
      }
      query += ") VALUES (";
      ncols = 0;
      for (auto& col : fColumns) {
         if (col.has_data) {
            if (ncols) query += ",";
            query += col.get_data_string_for_insert();
            ++ncols;
         }
      }
      query += ")";
      if (database::debug) std::cout << query << std::endl;
      fDB->execute(query);
   }
 
 
   //____________________________________________________________________________//
 
}