Merge branch 'development-v2' into development-v2-rscan-hotfix

Author: maxux

README.md

@@ -174,37 +174,41 @@ This mode is not possible if you don't have any data/index already available.
 
 # Supported commands
 - `PING`
-- `SET key value [timestamp]`
-- `GET key`
-- `MGET key`
-- `DEL key`
+- `SET <key> <value> [timestamp]`
+- `GET <key>`
+- `MGET <key> [key ...]`
+- `DEL <key>`
 - `STOP` (used only for debugging, to check memory leaks)
-- `EXISTS key`
-- `CHECK key`
-- `KEYCUR key`
+- `EXISTS <key>`
+- `CHECK <key>`
+- `KEYCUR <key>`
 - `INFO`
-- `NSNEW namespace`
-- `NSDEL namespace`
-- `NSINFO namespace`
+- `NSNEW <namespace>`
+- `NSDEL <namespace>`
+- `NSINFO <namespace>`
 - `NSLIST`
-- `NSSET namespace property value`
+- `NSSET <namespace> <property> <value>`
 - `NSJUMP`
-- `SELECT namespace [SECURE password]`
+- `SELECT <namespace> [SECURE password]`
 - `DBSIZE`
 - `TIME`
-- `AUTH password`
-- `AUTH SECURE`
-- `SCAN [optional cursor]`
-- `SCANX [optional cursor]` (this is just an alias for `SCAN`)
-- `RSCAN [optional cursor]`
+- `AUTH [password]`
+- `AUTH SECURE [password]`
+- `SCAN [cursor]`
+- `RSCAN [cursor]`
 - `WAIT command | * [timeout-ms]`
-- `HISTORY key [binary-data]`
+- `HISTORY <key> [binary-data]`
 - `FLUSH`
 - `HOOKS`
 - `INDEX DIRTY [RESET]`
+- `DATA RAW <fileid> <offset>`
+- `LENGTH <key>`
+- `KEYTIME <key>`
 
 `SET`, `GET` and `DEL`, `SCAN` and `RSCAN` supports binary keys.
 
+Arguments `<flags>` are mandatory, arguments `[flags]` are optionals.
+
 > Compared to real redis protocol, during a `SET`, the key is returned as response.
 
 ## SET
@@ -513,6 +517,59 @@ List the current namespace index files id which were modified since last reset
 ### INDEX DIRTY RESET
 Reset the dirty list
 
+## DATA
+
+This command have small internal operation on raw data file.
+
+### DATA RAW
+
+An internal call allows to retreive a raw data object from database only based on fileid and offset.
+This method of access should only be made by a valid fileid and offset, some protection are in
+place to avoid issues on wrong offset but not fully tested yet.
+    
+This command (only for admin) returns an array with the object requested:
+  1. Key
+  2. Previous Offset
+  3. Integrity CRC32
+  4. Flags
+  5. Timestamp
+  6. Payload
+
+In addition with NSINFO, this command can be used to query a database based on fielid/offset
+from another database used eg, for replication.
+
+You can use fields `data_current_id` and `data_current_offset` from `NSINFO` to query valid offsets.
+
+## LENGTH
+
+Returns payload size of a key or `(nil)` if not found.
+
+```
+> SET hello world
+"hello"
+
+> LENGTH hello
+(integer) 5
+
+> LENGTH notfound
+(nil)
+```
+
+## KEYTIME
+
+Return last-update timestamp of a key or `(nil)` if not found.
+
+```
+> SET hello world
+"hello"
+
+> KEYTIME hello
+(integer) 1664996517
+
+> KEYTIME notfound
+(nil)
+```
+
 # Namespaces
 A namespace is a dedicated directory on index and data root directory.
 A namespace is a complete set of key/data. Each namespace can be optionally protected by a password

libzdb/api.c

@@ -401,8 +401,10 @@ zdb_api_t *zdb_api_del(namespace_t *ns, void *key, size_t ksize) {
         return zdb_api_reply(ZDB_API_DELETED, NULL);
     }
 
+    time_t timestamp = time(NULL);
+
     // update data file, flag entry deleted
-    if(!data_delete(ns->data, entry->id, entry->idlength)) {
+    if(!data_delete(ns->data, entry->id, entry->idlength, timestamp)) {
         zdb_debug("[-] api: del: deleting data failed\n");
         return zdb_api_reply(ZDB_API_INTERNAL_ERROR, NULL);
     }

libzdb/data.c

@@ -305,6 +305,86 @@ static void data_open_final(data_root_t *root) {
     zdb_verbose("[+] data: active file: %s\n", root->datafile);
 }
 
+data_raw_t data_raw_get_real(int fd, off_t offset) {
+    data_raw_t raw;
+
+    memset(&raw, 0x00, sizeof(raw));
+
+    // moving to the header offset
+    lseek(fd, offset, SEEK_SET);
+
+    // 1. fetch fixed header object
+    // 2. that fixed header first byte contains id length
+    // 3. fetch id with correct length (from 2)
+    // 4. fetch payload with correct length (from 2)
+    // 5. TODO check checksum ?
+
+    zdb_debug("[+] data: raw: fetching header from offset\n");
+    if(read(fd, &raw.header, sizeof(data_entry_header_t)) != sizeof(data_entry_header_t)) {
+        zdb_warnp("data: raw: incorrect data header read");
+        return raw;
+    }
+
+    if(raw.header.datalength > ZDB_DATA_MAX_PAYLOAD) {
+        zdb_verbose("[-] data: raw: datalength from header too large\n");
+        return raw;
+    }
+
+    zdb_debug("[+] data: raw: fetching id from header offset\n");
+    if(!(raw.id = malloc(raw.header.idlength))) {
+        zdb_warnp("data: raw: id: malloc");
+        return raw;
+    }
+
+    if(read(fd, raw.id, raw.header.idlength) != raw.header.idlength) {
+        zdb_warnp("data: raw: incorrect id read");
+        return raw;
+    }
+
+    if(raw.header.flags & DATA_ENTRY_DELETED) {
+        zdb_debug("[+] data: raw: entry deleted\n");
+        return raw;
+    }
+
+    zdb_debug("[+] data: raw: fetching payload from header offset\n");
+    if(!(raw.payload.buffer = malloc(raw.header.datalength))) {
+        zdb_warnp("data: raw: payload: malloc");
+        return raw;
+    }
+
+    if(read(fd, raw.payload.buffer, raw.header.datalength) != raw.header.datalength) {
+        zdb_warnp("data: raw: incorrect payload read");
+        return raw;
+    }
+
+    // this validate return object to be valid
+    raw.payload.length = raw.header.datalength;
+
+    return raw;
+}
+
+// fetch data full object from specific offset
+data_raw_t data_raw_get(data_root_t *root, fileid_t dataid, off_t offset) {
+    int fd;
+    data_raw_t raw;
+
+    // initialize everything
+    memset(&raw, 0x00, sizeof(raw));
+
+    zdb_debug("[+] data: raw request: id %u, offset %lu\n", dataid, offset);
+
+    // acquire data id fd
+    if((fd = data_grab_dataid(root, dataid)) < 0)
+        return raw;
+
+    raw = data_raw_get_real(fd, offset);
+
+    // release dataid
+    data_release_dataid(root, dataid, fd);
+
+    return raw;
+}
+
 // jumping to the next id close the current data file
 // and open the next id file, it will create the new file
 size_t data_jump_next(data_root_t *root, fileid_t newid) {
@@ -547,7 +627,7 @@ int data_entry_is_deleted(data_entry_header_t *entry) {
 // was deleted
 // this is needed in order to rebuild an index from data file and
 // for compaction process
-int data_delete(data_root_t *root, void *id, uint8_t idlength) {
+int data_delete(data_root_t *root, void *id, uint8_t idlength, time_t timestamp) {
     unsigned char *empty = (unsigned char *) "";
 
     data_request_t dreq = {
@@ -557,6 +637,7 @@ int data_delete(data_root_t *root, void *id, uint8_t idlength) {
         .idlength = idlength,
         .flags = DATA_ENTRY_DELETED,
         .crc = 0,
+        .timestamp = timestamp,
     };
 
     zdb_debug("[+] data: delete: insert empty flagged data\n");

libzdb/data.h

@@ -3,6 +3,7 @@
 
     // split datafile after 256 MB
     #define ZDB_DEFAULT_DATA_MAXSIZE  256 * 1024 * 1024
+    #define ZDB_DATA_MAX_PAYLOAD      8 * 1024 * 1024
 
     // data statistics
     typedef struct data_stats_t {
@@ -73,6 +74,13 @@
 
     } data_payload_t;
 
+    typedef struct data_raw_t {
+        data_entry_header_t header;
+        uint8_t *id;
+        data_payload_t payload;
+
+    } data_raw_t;
+
     // scan internal representation
     // we use a status and a pointer to the header
     // in order to know what to do
@@ -124,12 +132,13 @@
     fileid_t data_dataid(data_root_t *root);
     void data_delete_files(char *datadir);
 
+    data_raw_t data_raw_get(data_root_t *root, fileid_t dataid, off_t offset);
     data_payload_t data_get(data_root_t *root, size_t offset, size_t length, fileid_t dataid, uint8_t idlength);
     int data_check(data_root_t *root, size_t offset, fileid_t dataid);
 
     // size_t data_match(data_root_t *root, void *id, uint8_t idlength, size_t offset, fileid_t dataid);
 
-    int data_delete(data_root_t *root, void *id, uint8_t idlength);
+    int data_delete(data_root_t *root, void *id, uint8_t idlength, time_t timestamp);
 
     // size_t data_insert(data_root_t *root, unsigned char *data, uint32_t datalength, void *vid, uint8_t idlength, uint8_t flags);
     size_t data_insert(data_root_t *root, data_request_t *source);

libzdb/index_loader.c

@@ -331,6 +331,7 @@ static size_t index_load_file(index_root_t *root) {
             .offset = entry->offset,
             .flags = entry->flags,
             .idxoffset = offset,
+            .timestamp = entry->timestamp,
             .crc = entry->crc,
             .parentid = entry->parentid,
             .parentoff = entry->parentoff,

tools/incremental-update/incremental.py

@@ -0,0 +1,63 @@
+import redis
+import time
+
+class ZDBIncremental:
+    def __init__(self, master, slave):
+        self.master = redis.Redis(unix_socket_path="/tmp/zdb.sock")
+        self.slave = redis.Redis(port=9900)
+
+        # disable defaults callbacks
+        for target in [self.master, self.slave]:
+            target.set_response_callback("NSINFO", redis.client.parse_info)
+            target.set_response_callback("DEL", redis.client.bool_ok)
+            target.set_response_callback("SET", bytes)
+
+    def sync(self, master, slave):
+        raw = self.master.execute_command("DATA", "RAW", slave['dataid'], slave['offset'])
+
+        print(raw)
+
+        if raw[3] == 0:
+            # SET
+            response = self.slave.execute_command("SET", raw[0], raw[5], raw[4])
+            if response != raw[0]:
+                raise RuntimeError(f"incorrect set {response}")
+
+        else:
+            # DEL
+            self.slave.execute_command("DEL", raw[0], raw[4])
+
+
+    def run(self):
+        namespace = "default"
+
+        while True:
+            master = {}
+            slave = {}
+
+            nsmaster = self.master.execute_command("NSINFO", namespace)
+            master['dataid'] = int(nsmaster['data_current_id'])
+            master['offset'] = int(nsmaster['data_current_offset'])
+
+            nsslave = self.slave.execute_command("NSINFO", namespace)
+            slave['dataid'] = int(nsslave['data_current_id'])
+            slave['offset'] = int(nsslave['data_current_offset'])
+
+            print("master: %d:%d" % (master['dataid'], master['offset']))
+            print("slave: %d:%d" % (slave['dataid'], slave['offset']))
+
+            if slave['dataid'] > master['dataid']:
+                raise RuntimeError("slave ahead from master")
+
+            if master['dataid'] == slave['dataid']:
+                if master['offset'] == slave['offset']:
+                    print("sync, waiting")
+                    time.sleep(10)
+                    continue
+
+            self.sync(master, slave)
+
+
+if __name__ == '__main__':
+    incremental = ZDBIncremental("hello", "world")
+    incremental.run()

zdbd/commands.c

@@ -121,6 +121,7 @@ static command_t commands_handlers[] = {
     {.command = "AUTH",    .handler = command_auth},     // custom AUTH command to authentifcate admin
     {.command = "HOOKS",   .handler = command_hooks},    // custom HOOKS command to list running hooks
     {.command = "INDEX",   .handler = command_index},    // custom INDEX command to query internal index
+    {.command = "DATA",    .handler = command_data},     // custom DATA command to query internal data
 
     // dataset
     {.command = "SET",     .handler = command_set},      // default SET command
@@ -136,6 +137,8 @@ static command_t commands_handlers[] = {
     {.command = "KSCAN",   .handler = command_kscan},    // custom command to iterate over keys matching pattern
     {.command = "HISTORY", .handler = command_history},  // custom command to get previous version of a key
     {.command = "KEYCUR",  .handler = command_keycur},   // custom command to get cursor id from a key
+    {.command = "LENGTH",  .handler = command_length},   // custom command to get value length of a key
+    {.command = "KEYTIME", .handler = command_keytime},  // custom command to get last updated timestamp of a key
 
     // query
     {.command = "INFO",    .handler = command_info},     // returns 0-db server name

zdbd/commands.h

@@ -21,4 +21,8 @@
 
     int command_error_locked(redis_client_t *client);
     int command_error_frozen(redis_client_t *client);
+
+    // defined in commands_set.c
+    // used by commands_set.c and commands_dataset.c
+    time_t timestamp_from_set(resp_request_t *request, int field);
 #endif