RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT > Class Template Reference

Template class for restoring an aggregate classe. More...

#include "rdm_time_series_restore_aggregate.h"

Inheritance diagram for RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >:

Collaboration diagram for RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >:

Public Member Functions
uint8_t *	serialize (uint8_t *buffer)
uint8_t *	unserialize (uint8_t *buffer)
RDM_RETCODE	init (RDM_DB p_db)
	Initialize this object. More...
void	reset (void)
	Reset this object. More...
template<class SOURCE_VALUE_T >
RDM_RETCODE	put_value (SOURCE_VALUE_T *psource_value, RDM::DB::TRANSACTIONAL_T=RDM::DB::NOT_TRANSACTIONAL)
	Template method for receiving a data value. More...
template<class SOURCE_STATS_T >
RDM_RETCODE	put_stats (SOURCE_STATS_T *psource_stats, RDM::DB::TRANSACTIONAL_T=RDM::DB::NOT_TRANSACTIONAL)
	Template method for receiving statistics. More...
template<class SOURCE_RANGE_T >
RDM_RETCODE	put_range (SOURCE_RANGE_T *psource_range, RDM::DB::TRANSACTIONAL_T=RDM::DB::NOT_TRANSACTIONAL)
	Template method for receiving ranges. More...
Public Member Functions inherited from RDM::TIME_SERIES::custom< NEXT >
uint8_t *	serialize (uint8_t *buffer)
uint8_t *	unserialize (uint8_t *buffer)
RDM_RETCODE	init (RDM_DB db)
	Initialize this object. More...
void	reset (void)
	Reset this object. More...
RDM_RETCODE	flush_value (uint32_t threshold=1, RDM::DB::TRANSACTIONAL_T transactional=RDM::DB::NOT_TRANSACTIONAL)
	Flush this object. More...
RDM_RETCODE	flush_stats (uint32_t threshold=1, RDM::DB::TRANSACTIONAL_T transactional=RDM::DB::NOT_TRANSACTIONAL)
	Flush this object. More...
RDM_RETCODE	flush_range (uint32_t threshold=1, RDM::DB::TRANSACTIONAL_T transactional=RDM::DB::NOT_TRANSACTIONAL)
	Flush this object. More...
template<class SOURCE_VALUE_T >
RDM_RETCODE	put_value (SOURCE_VALUE_T *source_value, RDM::DB::TRANSACTIONAL_T transactional=RDM::DB::NOT_TRANSACTIONAL)
	Template method for receiving a data value. More...
template<class SOURCE_STATS_T >
RDM_RETCODE	put_stats (SOURCE_STATS_T *source_stats, RDM::DB::TRANSACTIONAL_T transactional=RDM::DB::NOT_TRANSACTIONAL)
	Template method for receiving statistics. More...
template<class SOURCE_RANGE_T >
RDM_RETCODE	put_range (SOURCE_RANGE_T *source_range, RDM::DB::TRANSACTIONAL_T transactional=RDM::DB::NOT_TRANSACTIONAL)
	Template method for receiving ranges. More...

Static Public Member Functions
constexpr static int	get_serialize_size (void)
	Get the size of a buffer for serialization to hold its state. More...
Static Public Member Functions inherited from RDM::TIME_SERIES::custom< NEXT >
constexpr static int	get_serialize_size (void)
	Get the size of a buffer for serialization to hold its state. More...

Additional Inherited Members
Protected Member Functions inherited from RDM::TIME_SERIES::custom< NEXT >
uint32_t	init_tables_to_write_lock (RDM_TABLE_ID *tables)
	IDs of the tables where rows are inserted. More...
uint32_t	init_tables_to_read_lock (RDM_TABLE_ID *tables)
	IDs of the tables where rows are read. More...
void	unput (void)
	Undo a previous put operation. More...
void	unflush (void)
	Undo a previous flush operation. More...
Static Protected Member Functions inherited from RDM::TIME_SERIES::custom< NEXT >
constexpr static int	number_of_tables_to_write_lock (void)
	Number of tables where rows are inserted. More...
constexpr static int	number_of_tables_to_read_lock (void)
	Number of tables where rows are read. More...

Detailed Description

template<class TIME_STAMP_KEY_T, RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT>
class RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >

Template class for restoring an aggregate classe.

This template class can be used to restore the state of an aggregate class.

Before we explain exactly how this class works, let's consider the following scenario:

split <insert_row <TABLE_SOURCE>
 stats <32, AGGREGATE, insert_row <TABLE_AGGREGATE>>> ts;

When data is fed into the above class chain using ts.put_data(), the data will be inserted into the SOURCE table, and for every 32 rows inserted into the first table, a row with statistics will be inserted into the AGGREGATE table. However, if the program where this code is deployed is shut down and restarted, any rows left over in the aggregation (the stats class) after the last statistics has been inserted will not get counted towards any statistics.

Calling ts.flush_value() before shutdown can force statistics to be generated for the left over data. However, this is not always desireable or possible. For example, the RaimaDB Fast Fourier Transform (FFT) APIs classes are not able to honor a flush. The same is the case for the downsample class.

This class can remedy this. It is able to feed the missing data into the aggregation class (stats) when started up. If set up correctly, it will look up the last row from the AGGREGATE table and feed all the rows from the SOURCE table that have a timestamp greater than the timestamps in the AGGREGATE table into the stats class.

For this to work correctly, the above class chain has to be arranged slightly differently. The order of the two classes in the split has to be reversed. This class can then be inserted between the split and the stats class as follows:

split <restore_aggregate<UINT64_t,
 KEY_SOURCE_TIME_STAMP_CURRENT,
 KEY_AGGREGATE_TIME_STAMP_LAST,
stats <32, AGGREGATE,
 insert_row <TABLE_AGGREGATE>>>
 insert_row <TABLE_SOURCE>> ts;

This class can be used to restore the state for any aggregate class where it is sufficient to just feed some of the last original data to restore the state. For this to work, timestamp has to increase in value over time and the source and aggregate data must both be present.

It is even possible to restore the state for a chain of aggregate classes as long as there is access to the original data and the final aggregated data. Classes that do certain transformations can also be in the mix.

Template Parameters

TIME_STAMP_KEY_T	The type for the timestamp keys specified next. The type for SOURCE and the TARGET time stamps are required to be the same.
SOURCE_TIME_STAMP_KEY	The key ID to look up the timestamp for the SOURCE table. The key must be a key for the current or the last timestamp.
AGGREGATE_TIME_STAMP_KEY	The key ID to look up the timestamp for the AGGREGATE table. The key must be a key for the current or the last timestamp.
NEXT	The class in the chain to receive data

Examples

time_series_restore_aggregate.cpp.

Member Function Documentation

get_serialize_size()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

constexpr static int RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::get_serialize_size ( void )

inlinestaticconstexpr

Get the size of a buffer for serialization to hold its state.

Get the size needed for a buffer to hold the state of this object.

 {
return sizeof (restored_aggregate) + NEXT::get_serialize_size ();
 }

init()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

RDM_RETCODE RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::init ( RDM_DB p_db )

inline

Initialize this object.

Call this method before sending it any data values, ranges, or statistics.

Return values

sOKAY	Normal, successful return.
eINVARG	Invalid argument.
eDBNOTOPEN	Database not open.
eCURSORDB	Cursor is associated with a different database.
ePRECOMMITTED	A precommitted transaction must be committed or rolled back before further operations on this database are allowed.

Parameters

p_db	[IN] Use this database for chained classes that need to insert rows

 {
RDM_RETCODE rc = custom <NEXT>::init (p_db);


if (rc == sOKAY)
 {
 restored_aggregate = false;
 restore_aggregate::db = p_db;
 }


return rc;
 }

References sOKAY.

put_range()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

template<class SOURCE_RANGE_T >

RDM_RETCODE RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::put_range ( SOURCE_RANGE_T * psource_range, RDM::DB::TRANSACTIONAL_T = RDM::DB::NOT_TRANSACTIONAL )

inline

Template method for receiving ranges.

Receive one range and process it accordingly.

The type for the value provided is required to have columns for 'time_stamp_first', 'time_stamp_last', and 'value_range' with appropriate types. The column 'value_range' must be an array that is able to hold a static number of elements. Such a type can be generated using rdm-compile with a schema including a table similar to this:

CREATE TABLE range
(
 value_range DOUBLE ARRAY [32] NOT NULL,
 time_stamp_first UINT64 NOT NULL,
 time_stamp_last UINT64 PRIMARY KEY
);

This operation can be specified as being transactional. However, if the caller always rolls back the transaction in the case of a failure, there is no need to specify single operations to be transactional.

Template Parameters

SOURCE_RANGE_T

The actual type of the range received.

Return values

sOKAY	Normal, successful return.
eNOSTARTUPDATE	An update operation was attempted when no rdm_dbStartUpdate() is active.
eNOTLOCKED	Attempt to access a table for reading or update without proper locks.
eCURSORDB	Cursor is associated with a different database.
eDBNOTOPEN	Database not open.
ePRECOMMITTED	A precommitted transaction must be committed or rolled back before further operations on this database are allowed.
eDUPLICATE	Attempt to insert a duplicate value as a unique/primary key.
eREADONLY	Database is read-only and cannot be updated.
eREFINTEGRITY	Integrity constraint violation.
eROWLIMIT	Table row limit reached.
eINVARG	Invalid argument.

Parameters

[in]

psource_range

The source range sent to this class

 {
RDM_RETCODE rc = sOKAY;


if (restored_aggregate == false)
 {
RDM_CURSOR cursor = NULL;
 rc = rdm_dbGetRowsByKey (db, AGGREGATE_TIME_STAMP_KEY, &cursor);


if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToLast (cursor);
if (rc == sENDOFCURSOR)
 {
 rc = rdm_dbGetRowsByKey (db, SOURCE_TIME_STAMP_KEY, &cursor);
 }
else if (rc == sOKAY)
 {
 TIME_STAMP_KEY_T key;


 rc = rdm_cursorReadKey (cursor, AGGREGATE_TIME_STAMP_KEY, &key, sizeof (key), NULL);


if (rc == sOKAY)
 {
 rc = rdm_dbGetRowsByKey (db, SOURCE_TIME_STAMP_KEY, &cursor);
 }
if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToKey (cursor, SOURCE_TIME_STAMP_KEY, &key, sizeof (key));
 }
 }


if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToNext (cursor);
 }


while (rc == sOKAY)
 {
 SOURCE_RANGE_T source_range;
 rc = rdm_cursorReadRow (cursor, &source_range, sizeof (source_range), NULL);
if (rc == sOKAY)
 {
 rc = custom <NEXT>::put_range (&source_range);
 }
if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToNext (cursor);
 }
 }
if (rc == sENDOFCURSOR)
 {
 rc = sOKAY;
 }


rdm_cursorFree (cursor);
 }


if (rc == sOKAY)
 {
 restored_aggregate = true;
 }
 }


if (rc == sOKAY)
 {
 rc = custom <NEXT>::put_range (psource_range);
 }


return rc;
 }

References rdm_cursorFree(), rdm_cursorMoveToKey(), rdm_cursorMoveToLast(), rdm_cursorMoveToNext(), rdm_cursorReadKey(), rdm_cursorReadRow(), rdm_dbGetRowsByKey(), sENDOFCURSOR, and sOKAY.

Here is the call graph for this function:

put_stats()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

template<class SOURCE_STATS_T >

RDM_RETCODE RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::put_stats ( SOURCE_STATS_T * psource_stats, RDM::DB::TRANSACTIONAL_T = RDM::DB::NOT_TRANSACTIONAL )

inline

Template method for receiving statistics.

Receive one statistic and process it accordingly.

The type for the value provided is required to have columns for 'time_stamp_first', 'time_stamp_last', 'n', 'value_sum', and 'value_sum2' with appropriate types. Such a type can be generated using rdm-compile with a schema including a table similar to this:

CREATE TABLE stats
(
 time_stamp_first UINT64 NOT NULL,
 time_stamp_last UINT64 PRIMARY KEY,
 n UINT64 NOT NULL,
 value_sum DOUBLE NOT NULL,
 value_sum2 DOUBLE NOT NULL
);

This operation can be specified as being transactional. However, if the caller always rolls back the transaction in the case of a failure, there is no need to specify single operations to be transactional.

Template Parameters

SOURCE_STATS_T

The actual type of the statistic received.

Return values

sOKAY	Normal, successful return.
eNOSTARTUPDATE	An update operation was attempted when no rdm_dbStartUpdate() is active.
eNOTLOCKED	Attempt to access a table for reading or update without proper locks.
eCURSORDB	Cursor is associated with a different database.
eDBNOTOPEN	Database not open.
ePRECOMMITTED	A precommitted transaction must be committed or rolled back before further operations on this database are allowed.
eDUPLICATE	Attempt to insert a duplicate value as a unique/primary key.
eREADONLY	Database is read-only and cannot be updated.
eREFINTEGRITY	Integrity constraint violation.
eROWLIMIT	Table row limit reached.
eINVARG	Invalid argument.

Parameters

[in]

psource_stats

The source statistics sent to this class

 {
RDM_RETCODE rc = sOKAY;


if (restored_aggregate == false)
 {
RDM_CURSOR cursor = NULL;
 rc = rdm_dbGetRowsByKey (db, AGGREGATE_TIME_STAMP_KEY, &cursor);


if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToLast (cursor);
if (rc == sENDOFCURSOR)
 {
 rc = rdm_dbGetRowsByKey (db, SOURCE_TIME_STAMP_KEY, &cursor);
 }
else if (rc == sOKAY)
 {
 TIME_STAMP_KEY_T key;


 rc = rdm_cursorReadKey (cursor, AGGREGATE_TIME_STAMP_KEY, &key, sizeof (key), NULL);


if (rc == sOKAY)
 {
 rc = rdm_dbGetRowsByKey (db, SOURCE_TIME_STAMP_KEY, &cursor);
 }
if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToKey (cursor, SOURCE_TIME_STAMP_KEY, &key, sizeof (key));
 }
 }


if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToNext (cursor);
 }


while (rc == sOKAY)
 {
 SOURCE_STATS_T source_stats;
 rc = rdm_cursorReadRow (cursor, &source_stats, sizeof (source_stats), NULL);
if (rc == sOKAY)
 {
 rc = custom <NEXT>::put_stats (&source_stats);
 }
if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToNext (cursor);
 }
 }
if (rc == sENDOFCURSOR)
 {
 rc = sOKAY;
 }


rdm_cursorFree (cursor);
 }


if (rc == sOKAY)
 {
 restored_aggregate = true;
 }
 }


if (rc == sOKAY)
 {
 rc = custom <NEXT>::put_stats (psource_stats);
 }


return rc;
 }

References rdm_cursorFree(), rdm_cursorMoveToKey(), rdm_cursorMoveToLast(), rdm_cursorMoveToNext(), rdm_cursorReadKey(), rdm_cursorReadRow(), rdm_dbGetRowsByKey(), sENDOFCURSOR, and sOKAY.

Here is the call graph for this function:

put_value()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

template<class SOURCE_VALUE_T >

RDM_RETCODE RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::put_value ( SOURCE_VALUE_T * psource_value, RDM::DB::TRANSACTIONAL_T = RDM::DB::NOT_TRANSACTIONAL )

inline

Template method for receiving a data value.

Receive one data value and process it accordingly.

The type for the value provided is required to have columns for 'time_stamp_current' and 'value_current' with appropriate types. Such a type can be generated using rdm-compile with a schema including a table similar to this:

CREATE TABLE measurement
(
 time_stamp_current UINT64 PRIMARY KEY,
 value_current DOUBLE NOT NULL
);

This operation can be specified as being transactional. However, if the caller always rolls back the transaction in the case of a failure, there is no need to specify single operations to be transactional.

Template Parameters

SOURCE_VALUE_T

The actual type of the source values received

Return values

sOKAY	Normal, successful return.
eNOSTARTUPDATE	An update operation was attempted when no rdm_dbStartUpdate() is active.
eNOTLOCKED	Attempt to access a table for reading or update without proper locks.
eCURSORDB	Cursor is associated with a different database.
eDBNOTOPEN	Database not open.
ePRECOMMITTED	A precommitted transaction must be committed or rolled back before further operations on this database are allowed.
eDUPLICATE	Attempt to insert a duplicate value as a unique/primary key.
eREADONLY	Database is read-only and cannot be updated.
eREFINTEGRITY	Integrity constraint violation.
eROWLIMIT	Table row limit reached.
eINVARG	Invalid argument.

Parameters

[in]

psource_value

The source value sent to this class

 {
RDM_RETCODE rc = sOKAY;


if (restored_aggregate == false)
 {
RDM_CURSOR cursor = NULL;
 rc = rdm_dbGetRowsByKey (db, AGGREGATE_TIME_STAMP_KEY, &cursor);


if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToLast (cursor);
if (rc == sENDOFCURSOR)
 {
 rc = rdm_dbGetRowsByKey (db, SOURCE_TIME_STAMP_KEY, &cursor);
 }
else if (rc == sOKAY)
 {
 TIME_STAMP_KEY_T key;


 rc = rdm_cursorReadKey (cursor, AGGREGATE_TIME_STAMP_KEY, &key, sizeof (key), NULL);


if (rc == sOKAY)
 {
 rc = rdm_dbGetRowsByKey (db, SOURCE_TIME_STAMP_KEY, &cursor);
 }
if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToKey (cursor, SOURCE_TIME_STAMP_KEY, &key, sizeof (key));
 }
 }


if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToNext (cursor);
 }


while (rc == sOKAY)
 {
 SOURCE_VALUE_T source_value;
 rc = rdm_cursorReadRow (cursor, &source_value, sizeof (source_value), NULL);
if (rc == sOKAY)
 {
 rc = custom <NEXT>::put_value (&source_value);
 }
if (rc == sOKAY)
 {
 rc = rdm_cursorMoveToNext (cursor);
 }
 }
if (rc == sENDOFCURSOR)
 {
 rc = sOKAY;
 }


rdm_cursorFree (cursor);
 }


if (rc == sOKAY)
 {
 restored_aggregate = true;
 }
 }


if (rc == sOKAY)
 {
 rc = custom <NEXT>::put_value (psource_value);
 }


return rc;
 }

References rdm_cursorFree(), rdm_cursorMoveToKey(), rdm_cursorMoveToLast(), rdm_cursorMoveToNext(), rdm_cursorReadKey(), rdm_cursorReadRow(), rdm_dbGetRowsByKey(), sENDOFCURSOR, and sOKAY.

Here is the call graph for this function:

reset()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

void RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::reset ( void )

inline

Reset this object.

Calling this method to discard all internally kept data values, ranges, and statistics that have not yet been persisted and start over from scratch.

Any alternative implementation of this method in a sub class would call this method to reset the next class.

 {
custom <NEXT>::reset ();
 restored_aggregate = false;
 }

serialize()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

uint8_t* RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::serialize ( uint8_t * buffer )

inline

 {
 memcpy (buffer, &restored_aggregate, sizeof (restored_aggregate));
return custom <NEXT>::serialize (buffer + sizeof (restored_aggregate));
 }

unserialize()

template<class TIME_STAMP_KEY_T , RDM_KEY_ID SOURCE_TIME_STAMP_KEY, RDM_KEY_ID AGGREGATE_TIME_STAMP_KEY, class NEXT >

uint8_t* RDM::TIME_SERIES::restore_aggregate< TIME_STAMP_KEY_T, SOURCE_TIME_STAMP_KEY, AGGREGATE_TIME_STAMP_KEY, NEXT >::unserialize ( uint8_t * buffer )

inline

 {
 memcpy (&restored_aggregate, buffer, sizeof (restored_aggregate));
return custom <NEXT>::unserialize (buffer + sizeof (restored_aggregate));
 }

---

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

• rdm_time_series_restore_aggregate.h