Dynamic Schema Modification

With eXtremeDB version 7.1 and later it is possible to modify an SQL database schema through the standard SQL DDL statements create table/index, drop table/index and alter table/index. (Please refer to the eXtremeSQL User’s Guide for further details about SQL DDL.)

Replica-side replay of Dynamic Schema Modification

This enhancement stems from a request to register and propagate master-side CRUD events to the replica-side application when the database schema on the master is dynamically modified. This is accomplished by implementing an application-side iterator callback that lets the eXtremeDB High Availability database runtime publish information regarding the database modifications to the applications in JSON format. The iterator is available for all supported host languages, and as well can be enabled in the xSQL configuration file.

C API

The replica side mco_HA_replica_params_t is extended with two additional fields:

     
    mco_trans_iterator_callback_t iterator;
    void                         *iterator_context;
     

Where:

• Iterator is the application’s callback function that the database runtime calls when CRUD and some other events happen. The prototype for the callback is defined as the following in mco.h:

    typedef MCO_RET (*mco_trans_iterator_callback_t)(mco_trans_h trans,
    MCO_Hf* obj, int cid, int obj_state, void* user_ctx);
     

• iterator_context is the application's context passed into the callback through its last parameter user_ctx\

JSON output

Though applications can use the generic format for the callback function and implement custom output format, implemented and included into eXtremeDB High Availability distributions is an iterator that represents the database contents in JSON format (RFC 7159). The JSON converter synopsis as follows:

     
    mco_trans_iterator_h mco_create_json_converter(mco_stream_h stream,
    mco_json_converter_params_t *params);
     

The callback arguments are as follows:

• stream is a stream to write JSON values to. Must not be NULL
• params is an optional parameters struct as defined below (It can be NULL, in which case the default values described below are used:

The parameters struct is defined as follows:

 
    typedef struct {
        mco_bool         compact;
        mco_bool         ignore_stream_errors;
        MCO_RET          last_error;
        int              last_errno;
    } mco_json_converter_params_t;
     

Where:

• compact If set to true (default) a compact JSON is generated (with no spaces). false generates human-readable JSON format.
• ignore_stream_errors indicates whether to ignore errors while writing JSON to a stream. If set to false (default), any stream error causes the HA replication to stop-- the mco_HA_attach_master() returns MCO_E_WRITE_STREAM error and stopReason is set to MCO_HA_REPLICA_ITERATOR_ERROR. If true, errors are ignored and replication continues.
• last_error OUT: serves to return MCO_E_... errors from the mco_create_json_converter() function.
• last_errno OUT: serves to return an OS-level error (errno) from the mco_create_json_converter() function.

The optional parameters are initialized with default values by calling:

     
    void mco_json_converter_params_init(mco_json_converter_params_t *params);
     

If successful, the function has allocated and returned a pointer to the mco_trans_iterator_t structure (which is a common structure for all future iterator callback implementations) defined as follows:

     
    typedef struct mco_trans_iterator_t {
        mco_trans_iterator_callback_t callback;
        MCO_RET                       last_error;
        int                           last_errno;
    } mco_trans_iterator_t, *mco_trans_iterator_h;
     

Note that the mco_trans_iterator_t represents a header, similar to a base class in C++ terminology. Any custom iterator adds its own custom fields that are allocated beyond the header.

Following is an example:

 
    mco_trans_iterator_h json;
    mco_stream_h stream;
    mco_json_converter_params_t json_params; // declare parameters
    stream = ... <create stream, see below> ...
     
    void mco_json_converter_params_init(&json_params); // initialize parameters
    json_params.compact = true; // change default parameter value
    json = mco_create_json_converter(stream, &json_params); // create JSON iterator
     
    if (! json)
    {
        // Handle errors
    }
     
    mco_HA_replica_params_t replica_params;
    mco_HA_replica_params_init(&replica_params);
    replica_params.iterator         = json->callback; // set callback function
    replica_params.iterator_context = json;           // set context
    mco_HA_attach_master(db, conn_string, &replica_params, stop_reason, timeout);
     

Database objects rendered to the JSON object stream are separated by \0 (zero is not present in JSON). Therefore the stream can be processed as in the following code snippet (Java is used for simplicity):

     
    InputStream fis = (new java.net.Socket("127.0.0.1", 5566)).getInputStream();
    java.util.Scanner s = new java.util.Scanner(fis).useDelimiter("\\\0");
    while (s.hasNext())
    {
        JSONObject obj = new JSONObject(s.next());
        .....
    }
     

Each JSON object includes an operation field, which can be assigned one of the following values:

• TRANS_BEGIN – A transaction was started
• TRANS_END - Transaction was completed via a rollback or commit
• OBJ_NEW - A new object was created inside a transaction
• OBJ_UPDATE - An existing object was modified (inside a transaction)
• OBJ_DELETE - An object was deleted inside a transaction
• DELETE_ALL - All object from a class were erased (inside a transaction)
• DB_CLEAN - Master application executed mco_db_clean() outside a transaction
• SCHEMA_CHANGED - The database schema was modified outside a transaction

In addition JSON objects for OBJ_NEW, OBJ_UPDATE, OBJ_DELETE and DELETE_ALL operations include a table field. The value of the field indicates the table (class name) the affected object(s) belongs to. For the OBJ_NEW, OBJ_UPDATE and OBJ_DELETE operations an object field is included into the JSON object. The object field layout (the structure and data fields types) corresponds to the database object layout. Note the following:

1. Database arrays and vectors are represented via JSON arrays

2. Nested structures are represented as separate JSON objects

3. It is assumed that char and string database fields contain zero-terminated strings in the UTF-8 format

4. nchar and nstrings database field values are zero-terminated UTF-16 strings

5. wchar and wstring values are zero-terminated UTF-32 strings

6. BLOBs are represented as base64 JSON strings

7. NULL values and missing optional structure values are represented through the JSON -integrated literal null

For example, for the following SQL layout and content:

 
    create table a (i int, j int, s string);
    insert into a values (1,2,'test');
     

The following JSON would be generated:

     
    {
    "operation":"SCHEMA_CHANGED"
    }
    {
    "operation":"TRANS_BEGIN"
    }
    {
        "operation":"OBJ_NEW",
        "table":"a",
        "object":{
        "auto_oid@":1,
        "i":1,
        "j":2,
        "s":"test"
        }
    }
    {
        "operation":"TRANS_END"
    }
     

Extended eXtremeDB Streams

The eXtremeDB runtime customarily uses streams to output the database content to an external media and to import data into the database from an external media or a storage device. The eXtremeDB export / import APIs, mco_db_save() and mco_db_load() and some other functions use read- and write-streams to import / export data. The iterator extends stream functions with the ability to retain and then return error information from the stream back to the calling application or database runtime. The stream is represented through the mco_stream_h structure defined as follows:

     
    typedef struct mco_stream_t {
        mco_stream_write writer;
        MCO_RET          (*close) (struct mco_stream_t *self);
        MCO_RET          last_error;
        int              last_errno;
    } mco_stream_t, *mco_stream_h;
     

Here:

• writer is a standard eXtremeDB stream_writer:

     
    typedef mco_size_sig_t(*mco_stream_write)(void* stream_handle,
                            const void* from, mco_size_t nbytes);
     

• close represents the stream destructor
• last_error are MCO_E_... errors
• last_errno are OS-level errno values

Currently implemented are file-based, server_socket - based, client_socket –based and tee_stream -based stream functions.

File-based stream

File streams are created with either of the following functions:

 
    mco_stream_h mco_create_mcofile_stream(const char *filename,
                            mco_file_stream_params_t *params);
     

or

 
    mco_stream_h mco_create_stdfile_stream(const char *filename,
                            mco_file_stream_params_t *params);
     

The first function makes use of the eXtremeDB fs library through the mco_file_h handle, while the second uses the standard I/O (stdio.h) FILE* handle. Currently the only required parameter is the filename, which indicates the file name to write to, it must not be NULL. There are no optimal parameters at present, and the stream parameters structure only includes fields where to return errors received from the stream:

 
    typedef struct {
        MCO_RET          last_error;
        int              last_errno;
    } mco_file_stream_params_t;
 

However in the future some additional optional parameters (such as encryption) are possible, so the following function is called to initialize them:

 
    void mco_file_stream_params_init(mco_file_stream_params_t *params);
     

Following is a usage example:

 
    mco_stream_h stream;
    mco_file_stream_params_t file_params;
    mco_file_stream_params_init(&file_params);
    stream = mco_create_stdfile_stream("myfile.txt", &file_params);
     

or

 
    mco_stream_h stream = mco_create_stdfile_stream("myfile.txt", 0);
     

Server-socket stream

The following function is called to create a server-socket stream:

     
    mco_stream_h mco_create_server_socket_stream(int port, mco_server_socket_stream_params_t *params);
     

The only required parameter is the port number to listen on. Optional parameters are defined by the following structure:

     
    typedef struct {
        timer_unit        write_timeout;
        mco_size_t        buffer_size;
        mco_size_t        max_clients;
        const char       *net_interface;
        mco_sock_params_t sock_params;
        MCO_RET           last_error;
        int               last_errno;
    } mco_server_socket_stream_params_t;
     

Where:

• write_timeout is the write socket timeout in milliseconds The default is 1000 (1 second)
• buffer_size indicates the size of the buffer accumulating data to pass to the socket send() function. In bytes, the default is 16*1024
• max_clients - the maximum number of concurrently connected clients. The default is 16
• net_interface - network interface address to listen on. The default is 0.0.0.0 -- all interfaces
• sock_params - socket parameters: type, domain, SSL, etc.,
• last_error - OUT: serves to return MCO_E_... errors from the mco_create_server_socket_stream() function
• last_errno - OUT: serves to return OS-level error (errno) from the mco_create_server_socket_stream() function

Client-socket stream

The following function is called to create a client-socket stream:

     
    mco_stream_h mco_create_client_socket_stream(const char *hostname, int port,
    mco_client_socket_stream_params_t *params);
     

The required arguments are hostname (the name and IP addresses) and port (the port number). Optional parameters are defined by the following structure:

     
    typedef struct {
        timer_unit        write_timeout;
        timer_unit        connect_timeout;
        int               connect_attempts;
        timer_unit        connect_interval;
        mco_bool          auto_reconnect;
        mco_size_t        buffer_size;
        mco_sock_params_t sock_params;
        MCO_RET           last_error;
        int               last_errno;
    } mco_client_socket_stream_params_t;
     

Where:

• write_timeout - write socket timeout in milliseconds The default is 1000 (1 second)
• connect_timeout - a timeout for the connect() function in milliseconds. The default is 2000 (2 seconds)
• connect_attempts - A number of attempts to call connect() until the connection has been established. By default 3
• connect_interval - how long to wait before attempting to connect() again after unsuccessful attempt. By default no wait –0
• buffer_size - indicates the size of the buffer accumulating data to pass to the socket send() function. In bytes, the default is 16*1024
• sock_params - socket parameters: type, domain, SSL, etc.,
• last_error - OUT: serves to return MCO_E_... errors from the mco_create_client_socket_stream() function
• last_errno - OUT: serves to return OS-level error (errno) from the mco_create_client_socket_stream() function

Default values are initialized by calling:

     
    void mco_client_socket_stream_params_init(mco_client_socket_stream_params_t *params);
     

Following is a usage example:

     
    mco_stream_h stream;
    mco_client_socket_stream_params_t stream_params;
    mco_client_socket_stream_params_init(&stream_params);
    stream_params.connect_timeout = 3000; // change connect timeout
    // connect to myhost.com:10023
    stream = mco_create_client_socket_stream("myhost.com", 10023, &stream_params);
     

or

     
    // connect to myhost.com:10023
    mco_stream_h stream = mco_create_client_socket_stream("myhost.com", 10023, 0);
     

Tee stream

A tee-stream allows the iterator to write into two separate output streams simultaneously. For example one of the output streams can be a socket, while the other could be a file. Tee streams can be nested.

The following function is called to create a tee-stream:

     
    mco_stream_h mco_create_tee_stream(mco_stream_h stream1,
    mco_stream_h stream2, mco_bool any_ok);
     

Where:

• stream1 – the first output stream
• stream2 - the second output stream
• any_ok - if set to true, the write is deemed successful if the data was written successfully into any of the output streams. If set to false, both writes must have succeeded

Following is a usage example:

     
    // create file stream
    mco_stream_h file_stream   = mco_create_stdfile_stream("file.txt", 0);
     
    // create socket stream
    mco_stream_h socket_stream = mco_create_client_socket_stream("myhost.com",
                                        10023, 0);
    // join file and socket stream
    mco_stream_h tee_stream    = mco_create_tee_stream(file_stream, socket_stream, false);
     
    // create JSON iterator writing to both file and socket
    mco_trans_iterator_h json  = mco_create_json_converter(tee_stream, 0);
     

Custom stream

It is possible to inherit the functionality of an existing stream implementations and create a custom stream handle mco_stream_h based on the existing stream_handle and stream_writer:

     
    mco_stream_h mco_create_custom_stream(void* stream_handle,
                    mco_stream_write output_stream_writer,
                    MCO_RET(*close)(void *));
     

Where:

• stream_handle –stream context
• stream_writer - the stream function
• close - stream destructor, can be NULL

Following is a usage example:

     
    static mco_size_sig_t file_writer(void* stream_handle, const void* from, mco_size_t nbytes)
    {
        FILE *f = (FILE *) stream_handle;
        return fwrite(from, 1, nbytes, f);
    }
         
    FILE *f = fopen(filename, "w");
    mco_stream_h stream = mco_create_custom_stream(f, file_writer,(MCO_RET (*)(void*)) fclose);
         

Notes

• In case of errors all the above mco_create_XXX_stream() functions return NULL and set the appropriate error codes in params->last_error and params_last_errno (if defined)
• If a stream is destroyed, all nested streams are destroyed as well. For example for the tee stream:

     
    MCO_RET mco_destroy_stream(mco_stream_h s);
     

• When the iterator is destroyed, all dependent resources (streams) are also destroyed:

     
    MCO_RET mco_destroy_trans_iterator(mco_trans_iterator_h t);
     

For example:

 
    // create file stream
    mco_stream_h file_stream   = mco_create_stdfile_stream("file.txt", 0);
     
    // create socket stream
    mco_stream_h socket_stream = mco_create_client_socket_stream("myhost.com", 10023, 0);
     
    // join file and socket stream
    mco_stream_h tee_stream    = mco_create_tee_stream(file_stream,
    socket_stream, false);
     
    // create JSON iterator writing to both file and socket
    mco_trans_iterator_h json  = mco_create_json_converter(tee_stream, 0);
     
    //... Use iterator ...
    mco_destroy_trans_iterator(json);
     
    // Don't call mco_destroy_stream() for file_stream, socket_stream, tee_stream
     

• The iterator and the mco_stream_h are implemented in the target/mcoiter library. This library makes use only of the eXtremeDB public API (mco.h, mcouda.h and mconet.h). In addition the standard I/O (stdio.h) is utilized for the mco_create_stdfile_stream(). The use of the stdio can be turned off through the pound-define MCO_STDIO_FILE_STREAM in the target/mcoiter/mcoiter.c file:

     
    #define MCO_STDIO_FILE_STREAM 0
     

 

Finally, the "new style" streams can also be used with the eXtremeDB API functions written for the "old style" streams. The following snippet illustrates how to connect to the server and export the database content using the mco_db_save() API and the "new style" socket stream:

     
    mco_stream_h socket_stream = mco_create_client_socket_stream("myhost.com", 10023, 0);
    if (socket_stream)
    {
        mco_db_save(socket_stream, socket_stream->writer, connection);
        mco_destroy_stream(socket_stream);
    }
     

Java API

The Java-based iterator makes use of the standard Java OutputStream to output database content. The iterators themselves are defined in the TransIterator class:

     
    public class TransIterator {
    public abstract static class Iterator
    {
        public void close();
    }
     
    public static class JsonConverter extends Iterator
    {
        public JsonConverter(java.io.OutputStream os, boolean compact,
                    boolean ignoreStreamErrors);
                     
    public JsonConverter(java.io.OutputStream os);
    }
}

Here the TransIterator.Iterator represents the base class for all interior types and TransIterator.JsonConverter implements the JSON-based iterator.

Following is an example for the JSON iterator writing into a file:

     
    ReplicaConnection con = new ReplicaConnection(db);
    ReplicaConnection.Parameters replParams = new ReplicaConnection.Parameters();
    replParams.iterator = new TransIterator.JsonConverter(new java.io.FileOutputStream("myfile.txt"), false, false);
    con.attachMaster(connectionString, replParams, CONNECT_TIMEOUT);
    replParams.iterator.close();
     

C# API

Similar to Java the standard .NET Framework input/output streams are used:

     
    public class TransIterator
    {
        public abstract class Iterator
        {
            public Iterator(Connection con);
            public void Close();
        }
        public class JsonIterator : Iterator
        {
            public JsonIterator(Connection con, System.IO.Stream s, bool compact,
            bool ignore_stream_errors);
            public JsonIterator(Connection con, System.IO.Stream s);
        }
    }
     

Following is an example for the JSON iterator writing into a file:

     
    ReplicaConnection con = new ReplicaConnection(db);
    ReplicaConnection.Parameters replParams = new ReplicaConnection.Parameters();
    replParams.Iterator = new TransIterator.JsonIterator(con, new System.IO.FileStream("myfile.txt", System.IO.FileMode.Create));
    con.AttachMaster(connectionString, replParams, CONNECT_TIMEOUT);
    replParams.Iterator.Close();
     

Python API

The Python API is a wrapper around the C language mco_stream_h. As such most Python methods and their parameters are the same as the corresponding C functions.

To create a file-stream:

 
    def create_file_stream(filename)
     

To create a Server (listen=True) and client stream (listen= False):

     
    def create_socket_stream(hostname, port, listen = False,
    write_timeout = -1, buffer_size = -1, max_clients = -1,
    connect_timeout = -1, connect_attempts = -1, c
    onnect_interval = -1, reconnect = -1, socket_domain = -1 )
     

Note that for the server type socket the hostname can be None. If the hostname is not equal to None, then it corresponds to the mco_server_socket_stream_params_t::net_interface. The buffer_size, write_timeout, socket_domain parameters are common for both the client and the server sockets, while the max_clients parameter is defined only for the server, and connect_timeout, connect_attempts, connect_interval, reconnect only for the client sockets.

Following is an example for the JSON/Tee stream usage:

     
    db = exdb.open_database(dbname="rpldb", dictionary=dict, is_disk=is_disk,
    log_params=log_params)
    sock_stream = exdb.create_socket_stream("myhost.com", 10023, listen=False,
    write_timeout=200);
    file_stream = exdb.create_file_stream("out1.txt");
    tee_stream  = exdb.create_tee_stream(sock_stream, file_stream);
    json_ierator = exdb.create_json_iterator(tee_stream);
    replcon = exdb.ReplicaConnection(db)
    params = exdb.ReplicaConnectionParameters()
    params.iterator = json_iterator;
    replcon.attachMaster(ha_replica_connstr, params, CONNECT_TIMEOUT):
    exdb.destroy_trans_iterator(params.iterator);
     

xSQL API

For xSQL the iterator is described in the configuration file in the form of a JSON object. Each iterator contains a type field, which currently must be set to json. Other fields and their values depend on the iterator type. For the json type the fields are stream, compact (optional, boolean) and ignore_stream_errors (optional, boolean).

The Stream is also defined in the configuration file. The type of the stream can be file, socket or tee. Other fields and their values depend on the stream type.

For type file:

     
    {
        type : "file",
        name : "myfile" # file name
    }
     
    *type: "socket", listen: true (server-side socket:
    {
        type : "socket",
        listen : true,
        port   : 10000, # listen port
     
        buffer_size : 1k,  # optional, mco_server_socket_stream_params_t::buffer_size
        max_clients : 3,   # optional, mco_server_socket_stream_params_t::max_clients
        net_interface : 192.168.0.3, # optional, mco_server_socket_stream_params_t::net_interface
        write_timeout : 1000,  # optional, mco_server_socket_stream_params_t::write_timeout
         
        sock_params : {   # optional
            domain : "inet" # other options are "local" or "sdp",
                        # mco_server_socket_stream_params_t::sock_params::domain
            mode   : ["nodelay"], # combination of "nodelay", "do_not_reuse_address",
                        # "non_blocking", "keep_alive" or
                        # "do_not_cloexec", mco_server_socket stream_params_t::sock_params::mode
            sndbuf : 16k, # mco_server_socket_stream_params_t::sock_params::sndbuf
            use_ssl : false, # apply "ssl_params" for this socket or not
        }
    }
     

For type socket, listen: false (client socket):

 
    {
        type : "socket",
        listen : false,
        hostname : "192.168.0.1" # host to connect to,
        port     : 10023 # port to connect
        auto_reconnect : true,  # optional, mco_client_socket_stream_params_t::auto_reconnect
        buffer_size : 1k,   # optional, mco_client_socket_stream_params_t::buffer_size
        connect_attempts : 3,   # optional, mco_client_socket_stream_params_t::connect_attempts
        connect_interval : 1000, # optional, mco_client_socket_stream_params_t::connect_attempts
        connect_timeout : 2000, # optional, mco_client_socket_stream_params_t::connect_timeout
        write_timeout : 500, # optional, mco_client_socket_stream_params_t::write_timeout
         
        sock_params : { optional,
        domain : "inet" # other options are "local" or "sdp",
                    # mco_server_socket_stream_params_t::sock_params::domain
        mode   : ["nodelay"], # combination of "nodelay", "do_not_reuse_address",
                    # "non_blocking", "keep_alive" or
                    # "do_not_cloexec", mco_server_socket_stream_params_t::sock_params::mode
        sndbuf : 16k, # mco_server_socket_stream_params_t::sock_params::sndbuf
        use_ssl : false, # apply "ssl_params" for this socket or not
        }
    }
     

For type tee:

     
    {
        type : "tee",
        stream1 : {
            ...
        }
        stream2 : {
            ...
        }
    }
     

For example, the following configuration defines a replica-side JSON iterator that writes the file called myfile.txt and into a socket 127.0.0.1:10023

     
    ha_params : {
        connection_strings : "127.0.0.1:10000",
         
        replica_params : {
            iterator : {
            type : "json",
            compact : false,
            stream : {
                type   : "tee",
                stream1 : {
                    type     : "socket",
                    listen   : false,
                    hostname : "127.0.0.1",
                    port     : 10023,
                },
     
                stream2 : {
                    type : "file",
                    name : "myfile.txt"
                }
             }
          }
       }
    }