HsSqlSchema Class Reference

Inheritance diagram for HsSqlSchema:

Collaboration diagram for HsSqlSchema:

Public Member Functions
	HsSqlSchema ()
	~HsSqlSchema ()
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 ()
int	close ()
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[])
Public Member Functions inherited from HsSchema
	HsSchema ()
virtual	~HsSchema ()

Public Attributes
SqlBase *	fSql = NULL
std::string	fTableName
std::vector< std::string >	fColumnNames
std::vector< std::string >	fColumnTypes
std::vector< bool >	fColumnInactive
std::unordered_map< std::string, size_t >	fColumnIndexCache
Public Attributes inherited from HsSchema
std::string	fEventName
time_t	fTimeFrom = 0
time_t	fTimeTo = 0
std::vector< HsSchemaEntry >	fVariables
std::vector< int >	fOffsets
size_t	fNumBytes = 0
int	fCountWriteUndersize = 0
int	fCountWriteOversize = 0
size_t	fWriteMaxSize = 0
size_t	fWriteMinSize = 0
bool	fDisabled = true

Private Attributes
int	fTableTransactionCount = 0

Static Private Attributes
static std::map< SqlBase *, int >	gfTransactionCount

Detailed Description

Definition at line 682 of file history_schema.cxx.

Constructor & Destructor Documentation

HsSqlSchema()

HsSqlSchema::HsSqlSchema ( )

inline

Definition at line 698 of file history_schema.cxx.

   {
      // empty
   }

~HsSqlSchema()

HsSqlSchema::~HsSqlSchema ( )

inline

Definition at line 703 of file history_schema.cxx.

   {
      assert(get_transaction_count() == 0);
   }

Here is the call graph for this function:

Member Function Documentation

close()

int HsSqlSchema::close ( )

inlinevirtual

Implements HsSchema.

Definition at line 716 of file history_schema.cxx.

{ return close_transaction(); }

Here is the call graph for this function:

close_transaction()

int HsSqlSchema::close_transaction

(

)

Definition at line 4135 of file history_schema.cxx.

{
   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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

find_column_index()

int HsSqlSchema::find_column_index

(

const std::string &

column_name

)

Definition at line 740 of file history_schema.cxx.

{
   // 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;
}

Here is the caller graph for this function:

flush_buffers()

int HsSqlSchema::flush_buffers ( )

inlinevirtual

Implements HsSchema.

Definition at line 715 of file history_schema.cxx.

{ return close_transaction(); }

Here is the call graph for this function:

get_transaction_count()

int HsSqlSchema::get_transaction_count

(

)

Definition at line 4589 of file history_schema.cxx.

                                       {
   if (!fSql || fSql->fTransactionPerTable) {
      return fTableTransactionCount;
   } else {
      return gfTransactionCount[fSql];
   }
}

Here is the caller graph for this function:

increment_transaction_count()

void HsSqlSchema::increment_transaction_count

(

)

Definition at line 4605 of file history_schema.cxx.

                                              {
   if (!fSql || fSql->fTransactionPerTable) {
      fTableTransactionCount++;
   } else {
      gfTransactionCount[fSql]++;
   }
}

Here is the caller graph for this function:

match_event_var()

int HsSqlSchema::match_event_var ( const char *  event_name, const char *  var_name, const int  var_index  )

virtual

Reimplemented from HsSchema.

Definition at line 4167 of file history_schema.cxx.

{
   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);
}

Here is the call graph for this function:

print()

void HsSqlSchema::print ( bool  print_tags = true ) const

virtual

Reimplemented from HsSchema.

Definition at line 824 of file history_schema.cxx.

{
   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");
      }
   }
}

Here is the call graph for this function:

Here is the caller graph for this function:

read_data()

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[]  )

virtual

Implements HsSchema.

Definition at line 4500 of file history_schema.cxx.

{
   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;
}

Here is the call graph for this function:

read_last_written()

int HsSqlSchema::read_last_written ( const time_t  timestamp, const int  debug, time_t *  last_written  )

virtual

Implements HsSchema.

Definition at line 4453 of file history_schema.cxx.

{
   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;
}

Here is the call graph for this function:

remove_inactive_columns()

void HsSqlSchema::remove_inactive_columns ( )

virtual

Implements HsSchema.

Definition at line 4254 of file history_schema.cxx.

{
   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();
   }
}

reset_transaction_count()

void HsSqlSchema::reset_transaction_count

(

)

Definition at line 4597 of file history_schema.cxx.

                                          {
   if (!fSql || fSql->fTransactionPerTable) {
      fTableTransactionCount = 0;
   } else {
      gfTransactionCount[fSql] = 0;
   }
}

Here is the caller graph for this function:

write_event()

int HsSqlSchema::write_event ( const time_t  t, const char *  data, const size_t  data_size  )

virtual

Implements HsSchema.

Definition at line 4315 of file history_schema.cxx.

{
   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;
}

Here is the call graph for this function:

Member Data Documentation

fColumnInactive

std::vector<bool> HsSqlSchema::fColumnInactive

Definition at line 690 of file history_schema.cxx.

fColumnIndexCache

std::unordered_map<std::string, size_t> HsSqlSchema::fColumnIndexCache

Definition at line 694 of file history_schema.cxx.

fColumnNames

std::vector<std::string> HsSqlSchema::fColumnNames

Definition at line 688 of file history_schema.cxx.

fColumnTypes

std::vector<std::string> HsSqlSchema::fColumnTypes

Definition at line 689 of file history_schema.cxx.

fSql

SqlBase* HsSqlSchema::fSql = NULL

Definition at line 686 of file history_schema.cxx.

fTableName

std::string HsSqlSchema::fTableName

Definition at line 687 of file history_schema.cxx.

fTableTransactionCount

int HsSqlSchema::fTableTransactionCount = 0

private

Definition at line 734 of file history_schema.cxx.

gfTransactionCount

std::map< SqlBase *, int > HsSqlSchema::gfTransactionCount

staticprivate

Definition at line 735 of file history_schema.cxx.

---

The documentation for this class was generated from the following file:

• src/history_schema.cxx