From 3f1520c1b82c4275770e0f72a36029752ffd785b Mon Sep 17 00:00:00 2001 From: Admin <62771363+GoDeskio@users.noreply.github.com> Date: Sat, 31 Oct 2026 19:02:01 +0000 Subject: [PATCH] Add a loadable TS.* time-series module. Chunked Gorilla-style delta-of-delta and XOR compression, with range aggregation, label filters, retention, and RDB/AOF persistence. The compressor follows the published Gorilla paper and is not taken from another time-series product. Co-authored-by: Admin --- Dockerfile | 6 +- Dockerfile.modules | 5 + Makefile | 6 + modules/timeseries/.gitignore | 3 + modules/timeseries/Makefile | 39 ++ modules/timeseries/chunk.c | 612 ++++++++++++++++ modules/timeseries/chunk.h | 113 +++ modules/timeseries/chunk_test.c | 256 +++++++ modules/timeseries/query.c | 100 +++ modules/timeseries/ts.c | 1148 +++++++++++++++++++++++++++++++ tests/modules/timeseries.tcl | 141 ++++ tests/unit/timeseries.tcl | 3 + 12 files changed, 2431 insertions(+), 1 deletion(-) create mode 100644 modules/timeseries/.gitignore create mode 100644 modules/timeseries/Makefile create mode 100644 modules/timeseries/chunk.c create mode 100644 modules/timeseries/chunk.h create mode 100644 modules/timeseries/chunk_test.c create mode 100644 modules/timeseries/query.c create mode 100644 modules/timeseries/ts.c create mode 100644 tests/modules/timeseries.tcl create mode 100644 tests/unit/timeseries.tcl diff --git a/Dockerfile b/Dockerfile index 2d026c23d5c..fe47aae1eca 100644 --- a/Dockerfile +++ b/Dockerfile @@ -19,7 +19,10 @@ COPY . . # distclean drops any host object files that slipped into the build context. RUN make distclean || true \ && make -j"$(nproc)" BUILD_TLS=yes \ - && make install BUILD_TLS=yes PREFIX=/opt/cmo + && make install BUILD_TLS=yes PREFIX=/opt/cmo \ + && make -C modules/timeseries \ + && mkdir -p /opt/cmo/lib \ + && cp modules/timeseries/timeseries.so /opt/cmo/lib/timeseries.so FROM debian:bookworm-slim AS runtime @@ -37,6 +40,7 @@ RUN apt-get update \ && chmod 750 /data COPY --from=build /opt/cmo/bin/ /usr/local/bin/ +COPY --from=build /opt/cmo/lib/ /usr/local/lib/cmo/ COPY deploy/valkey.conf /etc/cache-me-outside/valkey.conf COPY deploy/docker-entrypoint.sh /usr/local/bin/docker-entrypoint.sh COPY deploy/healthcheck.sh /usr/local/bin/healthcheck.sh diff --git a/Dockerfile.modules b/Dockerfile.modules index e486d075af1..546ab56f1a1 100644 --- a/Dockerfile.modules +++ b/Dockerfile.modules @@ -29,3 +29,8 @@ FROM ${BASE} COPY --from=json /libjson.so /usr/local/lib/cmo-modules/libjson.so COPY --from=bloom /libbloom.so /usr/local/lib/cmo-modules/libbloom.so COPY --from=search /libsearch.so /usr/local/lib/cmo-modules/libsearch.so + +# From TS module PR +# Loadable modules for the Cache Me Outside image. +# The server image build runs this file's RUN lines from /src. +RUN make -C /src/modules/timeseries && mkdir -p /opt/cmo/lib && cp /src/modules/timeseries/timeseries.so /opt/cmo/lib/timeseries.so diff --git a/Makefile b/Makefile index 5c16c36ebaa..8742c3aaa75 100644 --- a/Makefile +++ b/Makefile @@ -2,6 +2,12 @@ default: all +# Forward every target into src/, then build loadable modules listed for the +# image. `all` is explicit so the module is part of a normal `make`. +all: + cd src && $(MAKE) all + $(MAKE) -C modules/timeseries + .DEFAULT: cd src && $(MAKE) $@ diff --git a/modules/timeseries/.gitignore b/modules/timeseries/.gitignore new file mode 100644 index 00000000000..8a7aff4fecc --- /dev/null +++ b/modules/timeseries/.gitignore @@ -0,0 +1,3 @@ +timeseries.so +chunk_test +*.o diff --git a/modules/timeseries/Makefile b/modules/timeseries/Makefile new file mode 100644 index 00000000000..18cb1091ed3 --- /dev/null +++ b/modules/timeseries/Makefile @@ -0,0 +1,39 @@ +# Time-series module. The .so is loaded with MODULE LOAD. chunk_test is a +# standalone compressor check and does not link the server. + +uname_S := $(shell sh -c 'uname -s 2>/dev/null || echo not') + +ifeq ($(uname_S),Linux) + SHOBJ_CFLAGS ?= -W -Wall -Wno-missing-field-initializers -fno-common -g -ggdb -std=gnu11 -O2 + SHOBJ_LDFLAGS ?= -shared + LD ?= gcc + CC ?= gcc +else + SHOBJ_CFLAGS ?= -W -Wall -Wno-missing-field-initializers -dynamic -fno-common -g -ggdb -std=gnu11 -O2 + SHOBJ_LDFLAGS ?= -bundle -undefined dynamic_lookup + LD ?= $(CC) +endif + +CFLAGS_MOD = -I. -I../../src $(CFLAGS) $(SHOBJ_CFLAGS) -fPIC + +.PHONY: all clean + +all: timeseries.so chunk_test + +chunk.o: chunk.c chunk.h + $(CC) $(CFLAGS_MOD) -c chunk.c -o chunk.o + +query.o: query.c chunk.h + $(CC) $(CFLAGS_MOD) -c query.c -o query.o + +ts.o: ts.c chunk.h + $(CC) $(CFLAGS_MOD) -c ts.c -o ts.o + +timeseries.so: ts.o chunk.o query.o + $(LD) -o $@ ts.o chunk.o query.o $(SHOBJ_LDFLAGS) $(LIBS) -lm + +chunk_test: chunk_test.c chunk.c chunk.h + $(CC) -I. -std=gnu11 -Wall -Wextra -O2 -g -o $@ chunk_test.c chunk.c -lm + +clean: + rm -f *.o timeseries.so chunk_test diff --git a/modules/timeseries/chunk.c b/modules/timeseries/chunk.c new file mode 100644 index 00000000000..bb066f716e1 --- /dev/null +++ b/modules/timeseries/chunk.c @@ -0,0 +1,612 @@ +/* Gorilla-style chunk compression. See chunk.h for the source of the scheme. */ +#include "chunk.h" + +#include +#include +#include + +static uint64_t double_bits(double v) { + uint64_t u; + memcpy(&u, &v, sizeof(u)); + return u; +} + +static double bits_double(uint64_t u) { + double v; + memcpy(&v, &u, sizeof(v)); + return v; +} + +static int ensure_bytes(tsChunk *c, size_t nbytes) { + if (nbytes <= c->alloc) + return 0; + size_t nalloc = c->alloc ? c->alloc : 64; + while (nalloc < nbytes) { + if (nalloc > (SIZE_MAX / 2)) + return -1; + nalloc *= 2; + } + unsigned char *grown = realloc(c->data, nalloc); + if (!grown) + return -1; + memset(grown + c->alloc, 0, nalloc - c->alloc); + c->data = grown; + c->alloc = nalloc; + return 0; +} + +/* Clamp the Gorilla leading-zero field to the 5 bits the paper uses. + * Extra leading zeros become part of the stored meaningful bits. */ +static void xor_shape(uint64_t xor, int *leading, int *trailing, + int *meaningful) { + int lead = (int)__builtin_clzll(xor); + int trail = (int)__builtin_ctzll(xor); + if (lead > 31) + lead = 31; + int meaning = 64 - lead - trail; + if (meaning < 1) + meaning = 1; + if (meaning > 64) + meaning = 64; + *leading = lead; + *trailing = trail; + *meaningful = meaning; +} + +static int put_bits(tsChunk *c, uint64_t value, int n) { + if (n <= 0) + return 0; + if (n > 64) + return -1; + size_t need = (size_t)(c->nbits + (uint32_t)n + 7) / 8; + if (ensure_bytes(c, need) != 0) + return -1; + for (int i = n - 1; i >= 0; i--) { + unsigned bit = (unsigned)((value >> i) & 1ull); + size_t byte = c->nbits / 8; + int shift = 7 - (int)(c->nbits % 8); + if (bit) + c->data[byte] |= (unsigned char)(1u << shift); + c->nbits++; + } + return 0; +} + +typedef struct bitr { + const unsigned char *data; + uint32_t nbits; + uint32_t pos; + int error; +} bitr; + +static uint64_t get_bits(bitr *r, int n) { + uint64_t value = 0; + if (n <= 0) + return 0; + if (r->pos + (uint32_t)n > r->nbits) { + r->error = 1; + return 0; + } + for (int i = 0; i < n; i++) { + size_t byte = r->pos / 8; + int shift = 7 - (int)(r->pos % 8); + unsigned bit = (r->data[byte] >> shift) & 1u; + value = (value << 1) | bit; + r->pos++; + } + return value; +} + +static int64_t sign_extend(uint64_t value, int bits) { + if (bits <= 0 || bits >= 64) + return (int64_t)value; + uint64_t sign = 1ull << (bits - 1); + if (value & sign) { + uint64_t mask = ~((1ull << bits) - 1ull); + return (int64_t)(value | mask); + } + return (int64_t)value; +} + +/* Bits required to append one sample, given the chunk's current predictor. */ +static int bits_for_sample(const tsChunk *c, int64_t ts, double value) { + int bits = 0; + if (!c->has_prev) + return 64 + 64; + int64_t delta = ts - c->prev_ts; + int64_t dod = delta - c->prev_delta; + if (dod == 0) + bits += 1; + else if (dod >= -64 && dod <= 63) + bits += 2 + 7; + else if (dod >= -256 && dod <= 255) + bits += 3 + 9; + else if (dod >= -2048 && dod <= 2047) + bits += 4 + 12; + else + bits += 4 + 64; + + uint64_t xor = double_bits(value) ^ c->prev_value_bits; + if (xor == 0) { + bits += 1; + return bits; + } + int leading, trailing, meaningful; + xor_shape(xor, &leading, &trailing, &meaningful); + if (c->has_value_window && leading >= (int)c->prev_leading && + trailing >= (int)c->prev_trailing) { + int window = 64 - (int)c->prev_leading - (int)c->prev_trailing; + bits += 2 + window; + } else { + bits += 2 + 5 + 6 + meaningful; + } + return bits; +} + +static int encode_dod(tsChunk *c, int64_t dod) { + if (dod == 0) + return put_bits(c, 0, 1); + if (dod >= -64 && dod <= 63) { + if (put_bits(c, 0x2, 2) != 0) + return -1; + return put_bits(c, (uint64_t)dod & 0x7f, 7); + } + if (dod >= -256 && dod <= 255) { + if (put_bits(c, 0x6, 3) != 0) + return -1; + return put_bits(c, (uint64_t)dod & 0x1ff, 9); + } + if (dod >= -2048 && dod <= 2047) { + if (put_bits(c, 0xe, 4) != 0) + return -1; + return put_bits(c, (uint64_t)dod & 0xfff, 12); + } + if (put_bits(c, 0xf, 4) != 0) + return -1; + return put_bits(c, (uint64_t)dod, 64); +} + +static int encode_xor(tsChunk *c, uint64_t bits) { + uint64_t xor = bits ^ c->prev_value_bits; + if (xor == 0) + return put_bits(c, 0, 1); + if (put_bits(c, 1, 1) != 0) + return -1; + int leading, trailing, meaningful; + xor_shape(xor, &leading, &trailing, &meaningful); + + if (c->has_value_window && leading >= (int)c->prev_leading && + trailing >= (int)c->prev_trailing) { + int window = 64 - (int)c->prev_leading - (int)c->prev_trailing; + if (window < 0) + window = 0; + if (window > 64) + window = 64; + if (put_bits(c, 0, 1) != 0) + return -1; + uint64_t field = c->prev_trailing >= 64 ? 0 : (xor >> c->prev_trailing); + return put_bits(c, field, window); + } + if (put_bits(c, 1, 1) != 0) + return -1; + if (put_bits(c, (uint64_t)leading, 5) != 0) + return -1; + /* 0 means 64 meaningful bits. meaningful is never 0 on this path. */ + if (put_bits(c, meaningful == 64 ? 0 : (uint64_t)meaningful, 6) != 0) + return -1; + uint64_t field = trailing >= 64 ? 0 : (xor >> trailing); + if (put_bits(c, field, meaningful) != 0) + return -1; + c->prev_leading = (uint8_t)leading; + c->prev_trailing = (uint8_t)trailing; + c->has_value_window = 1; + return 0; +} + +tsChunk *tsChunkCreate(size_t cap) { + if (cap < 16) + cap = 16; + tsChunk *c = calloc(1, sizeof(*c)); + if (!c) + return NULL; + c->data = calloc(1, cap); + if (!c->data) { + free(c); + return NULL; + } + c->cap = cap; + c->alloc = cap; + return c; +} + +void tsChunkFree(tsChunk *c) { + if (!c) + return; + free(c->data); + free(c); +} + +int tsChunkAppend(tsChunk *c, int64_t ts, double value) { + if (!c) + return -1; + if (c->count > 0) { + int need = bits_for_sample(c, ts, value); + size_t limit_bits = c->cap * 8; + /* The struct cap is the target chunk size. Grow only to finish a + * sample that was predicted to fit; a sample that would cross the + * target starts a new chunk instead. */ + if ((size_t)c->nbits + (size_t)need > limit_bits) + return 1; + } + if (!c->has_prev) { + if (put_bits(c, (uint64_t)ts, 64) != 0) + return -1; + if (put_bits(c, double_bits(value), 64) != 0) + return -1; + c->first_ts = ts; + c->last_ts = ts; + c->prev_ts = ts; + c->prev_delta = 0; + c->prev_value_bits = double_bits(value); + c->has_prev = 1; + c->count = 1; + return 0; + } + int64_t delta = ts - c->prev_ts; + int64_t dod = delta - c->prev_delta; + if (encode_dod(c, dod) != 0) + return -1; + if (encode_xor(c, double_bits(value)) != 0) + return -1; + c->prev_delta = delta; + c->prev_ts = ts; + c->prev_value_bits = double_bits(value); + c->last_ts = ts; + c->count++; + return 0; +} + +static int decode_one(bitr *r, tsChunk *state, int64_t *ts, double *value, + int first) { + if (first) { + *ts = (int64_t)get_bits(r, 64); + uint64_t bits = get_bits(r, 64); + if (r->error) + return -1; + *value = bits_double(bits); + state->prev_ts = *ts; + state->prev_delta = 0; + state->prev_value_bits = bits; + state->has_prev = 1; + return 0; + } + uint64_t marker = get_bits(r, 1); + int64_t dod; + if (marker == 0) { + dod = 0; + } else { + uint64_t marker2 = get_bits(r, 1); + if (marker2 == 0) { + dod = sign_extend(get_bits(r, 7), 7); + } else { + uint64_t marker3 = get_bits(r, 1); + if (marker3 == 0) { + dod = sign_extend(get_bits(r, 9), 9); + } else { + uint64_t marker4 = get_bits(r, 1); + if (marker4 == 0) + dod = sign_extend(get_bits(r, 12), 12); + else + dod = (int64_t)get_bits(r, 64); + } + } + } + if (r->error) + return -1; + int64_t delta = state->prev_delta + dod; + *ts = state->prev_ts + delta; + + uint64_t xor_flag = get_bits(r, 1); + uint64_t bits; + if (xor_flag == 0) { + bits = state->prev_value_bits; + } else { + uint64_t window_flag = get_bits(r, 1); + if (window_flag == 0) { + int window = 64 - (int)state->prev_leading - (int)state->prev_trailing; + if (window < 0) + window = 0; + uint64_t field = get_bits(r, window); + uint64_t xor + = state->prev_trailing >= 64 ? 0 : (field << state->prev_trailing); + bits = state->prev_value_bits ^ xor; + } else { + int leading = (int)get_bits(r, 5); + int meaningful = (int)get_bits(r, 6); + if (meaningful == 0) + meaningful = 64; + int trailing = 64 - leading - meaningful; + if (trailing < 0) { + r->error = 1; + return -1; + } + uint64_t field = get_bits(r, meaningful); + uint64_t xor = trailing >= 64 ? 0 : (field << trailing); + bits = state->prev_value_bits ^ xor; + state->prev_leading = (uint8_t)leading; + state->prev_trailing = (uint8_t)trailing; + state->has_value_window = 1; + } + } + if (r->error) + return -1; + *value = bits_double(bits); + state->prev_delta = delta; + state->prev_ts = *ts; + state->prev_value_bits = bits; + return 0; +} + +int tsChunkDecode(const tsChunk *c, tsSample **out, uint32_t *n) { + if (!c || !out || !n) + return -1; + *out = NULL; + *n = 0; + if (c->count == 0) + return 0; + tsSample *samples = malloc(sizeof(tsSample) * c->count); + if (!samples) + return -1; + bitr r = {.data = c->data, .nbits = c->nbits, .pos = 0, .error = 0}; + tsChunk state; + memset(&state, 0, sizeof(state)); + for (uint32_t i = 0; i < c->count; i++) { + if (decode_one(&r, &state, &samples[i].ts, &samples[i].value, i == 0) != + 0) { + free(samples); + return -1; + } + } + *out = samples; + *n = c->count; + return 0; +} + +int tsChunkRecompress(tsChunk *c, const tsSample *samples, uint32_t n) { + tsChunk *next = c->next; + size_t cap = c->cap ? c->cap : TS_CHUNK_DEFAULT_BYTES; + tsChunk *fresh = tsChunkCreate(cap); + if (!fresh) + return -1; + for (uint32_t i = 0; i < n; i++) { + int rc; + do { + rc = tsChunkAppend(fresh, samples[i].ts, samples[i].value); + if (rc == 1) { + if (fresh->cap > (SIZE_MAX / 2)) { + tsChunkFree(fresh); + return -1; + } + fresh->cap *= 2; + } + } while (rc == 1); + if (rc != 0) { + tsChunkFree(fresh); + return -1; + } + } + free(c->data); + *c = *fresh; + c->next = next; + free(fresh); + return 0; +} + +size_t tsChunkBytes(const tsChunk *c) { + if (!c) + return 0; + return sizeof(*c) + c->alloc; +} + +size_t tsSeriesStorageBytes(const tsSeries *s) { + size_t n = sizeof(*s); + if (!s) + return 0; + for (tsChunk *c = s->head; c; c = c->next) + n += tsChunkBytes(c); + n += (size_t)s->nlabels * sizeof(tsLabel); + for (int i = 0; i < s->nlabels; i++) { + if (s->labels[i].key) + n += strlen(s->labels[i].key) + 1; + if (s->labels[i].val) + n += strlen(s->labels[i].val) + 1; + } + if (s->key) + n += s->key_len + 1; + return n; +} + +int tsSeriesVisit(const tsSeries *s, + int (*fn)(const tsSample *sample, void *ud), void *ud) { + if (!s || !fn) + return -1; + for (const tsChunk *c = s->head; c; c = c->next) { + tsSample *samples = NULL; + uint32_t n = 0; + if (tsChunkDecode(c, &samples, &n) != 0) + return -1; + for (uint32_t i = 0; i < n; i++) { + int rc = fn(&samples[i], ud); + if (rc != 0) { + free(samples); + return rc; + } + } + free(samples); + } + return 0; +} + +int tsAggregate(const tsSample *in, size_t n, int agg, int64_t bucket, + tsSample **out, size_t *out_n) { + if (!out || !out_n) + return -1; + *out = NULL; + *out_n = 0; + if (agg == TS_AGG_NONE || n == 0) { + if (n == 0) + return 0; + tsSample *copy = malloc(sizeof(tsSample) * n); + if (!copy) + return -1; + memcpy(copy, in, sizeof(tsSample) * n); + *out = copy; + *out_n = n; + return 0; + } + if (bucket <= 0) + return -1; + size_t cap = 16; + size_t count = 0; + tsSample *acc = malloc(sizeof(tsSample) * cap); + if (!acc) + return -1; + int open = 0; + int64_t bucket_ts = 0; + double sum = 0; + double extreme = 0; + double first = 0; + double last = 0; + uint64_t seen = 0; + + for (size_t i = 0; i < n; i++) { + int64_t b = in[i].ts - (in[i].ts % bucket); + /* int64 division truncates toward zero. Timestamps we accept are >= 0, + * so this matches the usual floor alignment. */ + if (!open || b != bucket_ts) { + if (open) { + if (count == cap) { + size_t ncap = cap * 2; + tsSample *grown = realloc(acc, sizeof(tsSample) * ncap); + if (!grown) { + free(acc); + return -1; + } + acc = grown; + cap = ncap; + } + acc[count].ts = bucket_ts; + if (agg == TS_AGG_AVG) + acc[count].value = seen ? sum / (double)seen : 0; + else if (agg == TS_AGG_SUM) + acc[count].value = sum; + else if (agg == TS_AGG_MIN || agg == TS_AGG_MAX) + acc[count].value = extreme; + else if (agg == TS_AGG_COUNT) + acc[count].value = (double)seen; + else if (agg == TS_AGG_FIRST) + acc[count].value = first; + else + acc[count].value = last; + count++; + } + open = 1; + bucket_ts = b; + sum = in[i].value; + extreme = in[i].value; + first = in[i].value; + last = in[i].value; + seen = 1; + } else { + sum += in[i].value; + if (agg == TS_AGG_MIN && in[i].value < extreme) + extreme = in[i].value; + if (agg == TS_AGG_MAX && in[i].value > extreme) + extreme = in[i].value; + last = in[i].value; + seen++; + } + } + if (open) { + if (count == cap) { + tsSample *grown = realloc(acc, sizeof(tsSample) * (cap + 1)); + if (!grown) { + free(acc); + return -1; + } + acc = grown; + } + acc[count].ts = bucket_ts; + if (agg == TS_AGG_AVG) + acc[count].value = seen ? sum / (double)seen : 0; + else if (agg == TS_AGG_SUM) + acc[count].value = sum; + else if (agg == TS_AGG_MIN || agg == TS_AGG_MAX) + acc[count].value = extreme; + else if (agg == TS_AGG_COUNT) + acc[count].value = (double)seen; + else if (agg == TS_AGG_FIRST) + acc[count].value = first; + else + acc[count].value = last; + count++; + } + *out = acc; + *out_n = count; + return 0; +} + +int tsAggFromName(const char *name) { + if (!name) + return -1; + if (!strcasecmp(name, "avg")) + return TS_AGG_AVG; + if (!strcasecmp(name, "sum")) + return TS_AGG_SUM; + if (!strcasecmp(name, "min")) + return TS_AGG_MIN; + if (!strcasecmp(name, "max")) + return TS_AGG_MAX; + if (!strcasecmp(name, "count")) + return TS_AGG_COUNT; + if (!strcasecmp(name, "first")) + return TS_AGG_FIRST; + if (!strcasecmp(name, "last")) + return TS_AGG_LAST; + return -1; +} + +const char *tsDupPolicyName(int policy) { + switch (policy) { + case TS_DUP_FIRST: + return "FIRST"; + case TS_DUP_LAST: + return "LAST"; + case TS_DUP_MIN: + return "MIN"; + case TS_DUP_MAX: + return "MAX"; + case TS_DUP_SUM: + return "SUM"; + default: + return "BLOCK"; + } +} + +int tsDupPolicyFromName(const char *name) { + if (!name) + return -1; + if (!strcasecmp(name, "block")) + return TS_DUP_BLOCK; + if (!strcasecmp(name, "first")) + return TS_DUP_FIRST; + if (!strcasecmp(name, "last")) + return TS_DUP_LAST; + if (!strcasecmp(name, "min")) + return TS_DUP_MIN; + if (!strcasecmp(name, "max")) + return TS_DUP_MAX; + if (!strcasecmp(name, "sum")) + return TS_DUP_SUM; + return -1; +} diff --git a/modules/timeseries/chunk.h b/modules/timeseries/chunk.h new file mode 100644 index 00000000000..2826488ff8f --- /dev/null +++ b/modules/timeseries/chunk.h @@ -0,0 +1,113 @@ +/* Time-series chunks and the in-memory series object. + * + * Sample compression follows the published Gorilla scheme + * (Pelkonen et al., "Gorilla: A Fast, Scalable, In-Memory Time Series + * Database", SIGMOD 2015): delta-of-delta timestamps and XOR'd doubles. + * The bit layout is our own reading of that paper. A delta that does not + * fit in 32 bits is stored in 64 bits so a chunk can hold any millisecond + * timestamp. Nothing here is taken from another time-series product. + * + * Copyright (c) 2026, Cache Me Outside contributors. + * Licensed under the BSD 3-Clause License, the same terms as the server. + */ +#ifndef CMO_TS_CHUNK_H +#define CMO_TS_CHUNK_H + +#include +#include + +#define TS_CHUNK_DEFAULT_BYTES 4096 +#define TS_CHUNK_MIN_BYTES 128 +#define TS_CHUNK_MAX_BYTES 65536 + +enum { + TS_DUP_BLOCK = 0, + TS_DUP_FIRST = 1, + TS_DUP_LAST = 2, + TS_DUP_MIN = 3, + TS_DUP_MAX = 4, + TS_DUP_SUM = 5 +}; + +enum { + TS_AGG_NONE = 0, + TS_AGG_AVG = 1, + TS_AGG_SUM = 2, + TS_AGG_MIN = 3, + TS_AGG_MAX = 4, + TS_AGG_COUNT = 5, + TS_AGG_FIRST = 6, + TS_AGG_LAST = 7 +}; + +typedef struct tsSample { + int64_t ts; + double value; +} tsSample; + +typedef struct tsChunk tsChunk; + +struct tsChunk { + tsChunk *next; + int64_t first_ts; + int64_t last_ts; + uint32_t count; + size_t + cap; /* Target size. A sample that would pass this starts a new chunk. */ + size_t alloc; /* Bytes allocated for data. May exceed cap to finish one + sample. */ + uint32_t nbits; + unsigned char *data; + int64_t prev_ts; + int64_t prev_delta; + uint64_t prev_value_bits; + uint8_t prev_leading; + uint8_t prev_trailing; + int has_value_window; + int has_prev; +}; + +typedef struct tsLabel { + char *key; + char *val; +} tsLabel; + +typedef struct tsSeries tsSeries; + +struct tsSeries { + int64_t retention_ms; + int dup_policy; + size_t chunk_size; + uint64_t total_samples; + tsChunk *head; + tsChunk *tail; + int nlabels; + tsLabel *labels; + int64_t last_ts; + double last_value; + int has_sample; + char *key; + size_t key_len; +}; + +tsChunk *tsChunkCreate(size_t cap); +void tsChunkFree(tsChunk *c); +/* 0 appended, 1 chunk is full and the sample was not written, -1 error. */ +int tsChunkAppend(tsChunk *c, int64_t ts, double value); +int tsChunkDecode(const tsChunk *c, tsSample **out, uint32_t *n); +int tsChunkRecompress(tsChunk *c, const tsSample *samples, uint32_t n); +size_t tsChunkBytes(const tsChunk *c); + +size_t tsSeriesStorageBytes(const tsSeries *s); +int tsSeriesVisit(const tsSeries *s, + int (*fn)(const tsSample *sample, void *ud), void *ud); + +/* agg is a TS_AGG_* value. bucket must be > 0 when agg is not TS_AGG_NONE. + * On success *out is malloc'd and must be freed by the caller. */ +int tsAggregate(const tsSample *in, size_t n, int agg, int64_t bucket, + tsSample **out, size_t *out_n); +int tsAggFromName(const char *name); +const char *tsDupPolicyName(int policy); +int tsDupPolicyFromName(const char *name); + +#endif diff --git a/modules/timeseries/chunk_test.c b/modules/timeseries/chunk_test.c new file mode 100644 index 00000000000..b0ab5ecfd6e --- /dev/null +++ b/modules/timeseries/chunk_test.c @@ -0,0 +1,256 @@ +/* Compression round-trip, aggregation reference, and the 1M-point budget. + * Build: make -C modules/timeseries chunk_test && + * ./modules/timeseries/chunk_test + */ +#include "chunk.h" + +#include +#include +#include + +static int failures = 0; + +static void expect(int cond, const char *msg) { + if (!cond) { + fprintf(stderr, "FAIL %s\n", msg); + failures++; + } +} + +static int samples_equal(const tsSample *a, const tsSample *b) { + uint64_t ba, bb; + if (a->ts != b->ts) + return 0; + memcpy(&ba, &a->value, sizeof(ba)); + memcpy(&bb, &b->value, sizeof(bb)); + return ba == bb; +} + +static int round_trip(const tsSample *in, uint32_t n, size_t chunk_bytes) { + tsChunk *head = NULL; + tsChunk *tail = NULL; + for (uint32_t i = 0; i < n; i++) { + if (!tail) { + tail = tsChunkCreate(chunk_bytes); + head = tail; + if (!tail) + return -1; + } + int rc = tsChunkAppend(tail, in[i].ts, in[i].value); + if (rc == 1) { + tsChunk *next = tsChunkCreate(chunk_bytes); + if (!next) + return -1; + tail->next = next; + tail = next; + rc = tsChunkAppend(tail, in[i].ts, in[i].value); + } + if (rc != 0) + return -1; + } + uint32_t seen = 0; + for (tsChunk *c = head; c; c = c->next) { + tsSample *out = NULL; + uint32_t got = 0; + if (tsChunkDecode(c, &out, &got) != 0) + return -1; + for (uint32_t i = 0; i < got; i++) { + if (seen >= n || !samples_equal(&in[seen], &out[i])) { + fprintf(stderr, "mismatch at %u ts %lld\n", seen, + (long long)(seen < n ? in[seen].ts : -1)); + free(out); + return -1; + } + seen++; + } + free(out); + } + tsChunk *c = head; + while (c) { + tsChunk *next = c->next; + tsChunkFree(c); + c = next; + } + return seen == n ? 0 : -1; +} + +static void naive_avg(const tsSample *in, size_t n, int64_t bucket, + tsSample **out, size_t *out_n) { + size_t cap = 8, count = 0; + tsSample *acc = malloc(sizeof(tsSample) * cap); + int open = 0; + int64_t bts = 0; + double sum = 0; + uint64_t seen = 0; + for (size_t i = 0; i < n; i++) { + int64_t b = in[i].ts - (in[i].ts % bucket); + if (!open || b != bts) { + if (open) { + acc[count].ts = bts; + acc[count].value = sum / (double)seen; + count++; + if (count == cap) { + cap *= 2; + acc = realloc(acc, sizeof(tsSample) * cap); + } + } + open = 1; + bts = b; + sum = in[i].value; + seen = 1; + } else { + sum += in[i].value; + seen++; + } + } + if (open) { + acc[count].ts = bts; + acc[count].value = sum / (double)seen; + count++; + } + *out = acc; + *out_n = count; +} + +static uint64_t rng_state = 0x123456789abcdefULL; + +static uint64_t rnd(void) { + rng_state = rng_state * 6364136223846793005ULL + 1ULL; + return rng_state; +} + +int main(void) { + enum { N = 20000 }; + tsSample *in = malloc(sizeof(tsSample) * N); + int64_t ts = 0; + for (int i = 0; i < N; i++) { + ts += 1 + (int64_t)(rnd() % 1000); + /* Mix integers, fractions, and a run of identical values. */ + double v; + if ((i % 17) == 0) + v = (double)i; + else if ((i % 17) == 1) + v = (double)(i - 1); + else + v = (double)(int)(rnd() % 1000000) / 8.0; + in[i].ts = ts; + in[i].value = v; + } + expect(round_trip(in, N, 4096) == 0, "fuzz round trip"); + expect(round_trip(in, N, 128) == 0, "small chunk round trip"); + + tsSample *agg = NULL; + size_t agg_n = 0; + expect(tsAggregate(in, 1000, TS_AGG_AVG, 5000, &agg, &agg_n) == 0, + "aggregate"); + tsSample *naive = NULL; + size_t naive_n = 0; + naive_avg(in, 1000, 5000, &naive, &naive_n); + expect(agg_n == naive_n, "agg count matches naive"); + if (agg_n == naive_n) { + for (size_t i = 0; i < agg_n; i++) { + if (agg[i].ts != naive[i].ts || agg[i].value != naive[i].value) { + expect(0, "agg value matches naive"); + break; + } + } + } + free(agg); + free(naive); + + /* Other reducers on a hand-built series. */ + tsSample hand[6] = {{0, 1}, {10, 5}, {20, 3}, {100, 4}, {110, 8}, {150, 1}}; + tsSample *min_out = NULL; + size_t min_n = 0; + expect(tsAggregate(hand, 6, TS_AGG_MIN, 100, &min_out, &min_n) == 0, + "min agg"); + expect(min_n == 2, "min buckets"); + if (min_n == 2) { + expect(min_out[0].ts == 0 && min_out[0].value == 1, "min bucket 0"); + expect(min_out[1].ts == 100 && min_out[1].value == 1, "min bucket 1"); + } + free(min_out); + tsSample *sum_out = NULL; + size_t sum_n = 0; + expect(tsAggregate(hand, 6, TS_AGG_SUM, 100, &sum_out, &sum_n) == 0, + "sum agg"); + if (sum_n == 2) { + expect(sum_out[0].value == 9, "sum bucket 0"); + expect(sum_out[1].value == 13, "sum bucket 1"); + } + free(sum_out); + tsSample *cnt = NULL; + size_t cnt_n = 0; + expect(tsAggregate(hand, 6, TS_AGG_COUNT, 100, &cnt, &cnt_n) == 0, + "count agg"); + if (cnt_n == 2) { + expect(cnt[0].value == 3 && cnt[1].value == 3, "count values"); + } + free(cnt); + tsSample *first = NULL; + size_t first_n = 0; + expect(tsAggregate(hand, 6, TS_AGG_FIRST, 100, &first, &first_n) == 0, + "first agg"); + if (first_n == 2) + expect(first[0].value == 1 && first[1].value == 4, "first values"); + free(first); + tsSample *last = NULL; + size_t last_n = 0; + expect(tsAggregate(hand, 6, TS_AGG_LAST, 100, &last, &last_n) == 0, + "last agg"); + if (last_n == 2) + expect(last[0].value == 3 && last[1].value == 1, "last values"); + free(last); + + /* 1M smooth points must stay well under the 16-byte naive layout. */ + const uint32_t million = 1000000; + size_t bytes = 0; + uint32_t produced = 0; + tsChunk *head = NULL; + tsChunk *tail = NULL; + for (uint32_t i = 0; i < million; i++) { + if (!tail) { + tail = tsChunkCreate(4096); + head = tail; + } + int rc = tsChunkAppend(tail, (int64_t)i, (double)i * 0.1); + if (rc == 1) { + tsChunk *next = tsChunkCreate(4096); + tail->next = next; + tail = next; + rc = tsChunkAppend(tail, (int64_t)i, (double)i * 0.1); + } + if (rc != 0) { + expect(0, "1M append"); + break; + } + produced++; + } + for (tsChunk *c = head; c; c = c->next) + bytes += tsChunkBytes(c); + printf("1M points use %zu bytes (%.2f bytes/sample)\n", bytes, + produced ? (double)bytes / (double)produced : 0.0); + expect(bytes < 8u * 1024u * 1024u, "1M points under 8MB"); + /* Spot-check the compressed stream still decodes. */ + tsSample *spot = NULL; + uint32_t spot_n = 0; + expect(tsChunkDecode(head, &spot, &spot_n) == 0, "decode first chunk"); + if (spot_n > 0) + expect(spot[0].ts == 0 && spot[0].value == 0, "first sample"); + free(spot); + + tsChunk *c = head; + while (c) { + tsChunk *next = c->next; + tsChunkFree(c); + c = next; + } + free(in); + + if (failures) { + fprintf(stderr, "%d failure(s)\n", failures); + return 1; + } + printf("OK\n"); + return 0; +} diff --git a/modules/timeseries/query.c b/modules/timeseries/query.c new file mode 100644 index 00000000000..95ecc66a9e2 --- /dev/null +++ b/modules/timeseries/query.c @@ -0,0 +1,100 @@ +/* Range, reverse-range, and bucket aggregation replies. + * + * Aggregation is a straight bucket walk over decoded samples. The C test + * checks it against an independent naive implementation. */ +#include "chunk.h" + +#include "../src/valkeymodule.h" + +#include +#include + +typedef struct collect_state { + int64_t from; + int64_t to; + int filter_value; + double vmin; + double vmax; + tsSample *samples; + size_t n; + size_t cap; + int error; +} collect_state; + +static int collect_sample(const tsSample *sample, void *ud) { + collect_state *st = ud; + if (sample->ts < st->from || sample->ts > st->to) + return 0; + if (st->filter_value && + (sample->value < st->vmin || sample->value > st->vmax)) + return 0; + if (st->n == st->cap) { + size_t ncap = st->cap ? st->cap * 2 : 64; + tsSample *grown = realloc(st->samples, sizeof(tsSample) * ncap); + if (!grown) { + st->error = 1; + return -1; + } + st->samples = grown; + st->cap = ncap; + } + st->samples[st->n++] = *sample; + return 0; +} + +static void reply_sample(ValkeyModuleCtx *ctx, const tsSample *sample) { + ValkeyModule_ReplyWithArray(ctx, 2); + ValkeyModule_ReplyWithLongLong(ctx, (long long)sample->ts); + ValkeyModule_ReplyWithDouble(ctx, sample->value); +} + +/* Write the sample array for one series. Returns 0 on success. */ +int tsReplySamples(ValkeyModuleCtx *ctx, const tsSeries *s, int64_t from, + int64_t to, int reverse, int agg, int64_t bucket, + long long max_count, int filter_value, double vmin, + double vmax) { + collect_state st; + memset(&st, 0, sizeof(st)); + st.from = from; + st.to = to; + st.filter_value = filter_value; + st.vmin = vmin; + st.vmax = vmax; + if (s && tsSeriesVisit(s, collect_sample, &st) != 0) { + free(st.samples); + return -1; + } + if (st.error) { + free(st.samples); + return -1; + } + + tsSample *out = st.samples; + size_t n = st.n; + if (agg != TS_AGG_NONE) { + tsSample *agg_out = NULL; + size_t agg_n = 0; + if (tsAggregate(st.samples, st.n, agg, bucket, &agg_out, &agg_n) != 0) { + free(st.samples); + return -1; + } + free(st.samples); + out = agg_out; + n = agg_n; + } + if (reverse && n > 1) { + for (size_t i = 0; i < n / 2; i++) { + tsSample tmp = out[i]; + out[i] = out[n - 1 - i]; + out[n - 1 - i] = tmp; + } + } + if (max_count >= 0 && (size_t)max_count < n) + n = (size_t)max_count; + + ValkeyModule_ReplyWithArray(ctx, (long)n); + for (size_t i = 0; i < n; i++) + reply_sample(ctx, &out[i]); + free(out); + return 0; +} diff --git a/modules/timeseries/ts.c b/modules/timeseries/ts.c new file mode 100644 index 00000000000..d0437734723 --- /dev/null +++ b/modules/timeseries/ts.c @@ -0,0 +1,1148 @@ +/* TS.* time-series type. + * + * Command names follow the usual TS.* client API. Storage, compression, + * and query execution are implemented in this module. See chunk.h. + * + * Copyright (c) 2026, Cache Me Outside contributors. + * Licensed under the BSD 3-Clause License, the same terms as the server. + */ +#include "chunk.h" + +#include "../src/valkeymodule.h" + +#include +#include +#include +#include + +static ValkeyModuleType *TsType; + +int tsReplySamples(ValkeyModuleCtx *ctx, const tsSeries *s, int64_t from, + int64_t to, int reverse, int agg, int64_t bucket, + long long max_count, int filter_value, double vmin, + double vmax); + +static char *dup_cstr(const char *p, size_t len) { + char *s = malloc(len + 1); + if (!s) + return NULL; + memcpy(s, p, len); + s[len] = '\0'; + return s; +} + +static tsSeries *series_create(size_t chunk_size, int64_t retention, + int dup_policy) { + tsSeries *s = calloc(1, sizeof(*s)); + if (!s) + return NULL; + if (chunk_size < TS_CHUNK_MIN_BYTES) + chunk_size = TS_CHUNK_MIN_BYTES; + if (chunk_size > TS_CHUNK_MAX_BYTES) + chunk_size = TS_CHUNK_MAX_BYTES; + s->chunk_size = chunk_size; + s->retention_ms = retention; + s->dup_policy = dup_policy; + return s; +} + +static void series_clear_chunks(tsSeries *s) { + tsChunk *c = s->head; + while (c) { + tsChunk *next = c->next; + tsChunkFree(c); + c = next; + } + s->head = s->tail = NULL; + s->total_samples = 0; + s->has_sample = 0; +} + +static void series_free(tsSeries *s) { + if (!s) + return; + series_clear_chunks(s); + for (int i = 0; i < s->nlabels; i++) { + free(s->labels[i].key); + free(s->labels[i].val); + } + free(s->labels); + free(s->key); + free(s); +} + +static void series_set_key(tsSeries *s, const char *key, size_t len) { + char *copy = dup_cstr(key, len); + if (!copy) + return; + free(s->key); + s->key = copy; + s->key_len = len; +} + +static int series_set_label(tsSeries *s, const char *key, const char *val) { + for (int i = 0; i < s->nlabels; i++) { + if (strcmp(s->labels[i].key, key) == 0) { + char *copy = strdup(val); + if (!copy) + return -1; + free(s->labels[i].val); + s->labels[i].val = copy; + return 0; + } + } + tsLabel *grown = + realloc(s->labels, sizeof(tsLabel) * (size_t)(s->nlabels + 1)); + if (!grown) + return -1; + s->labels = grown; + s->labels[s->nlabels].key = strdup(key); + s->labels[s->nlabels].val = strdup(val); + if (!s->labels[s->nlabels].key || !s->labels[s->nlabels].val) + return -1; + s->nlabels++; + return 0; +} + +static int chunk_push_sample(tsSeries *s, int64_t ts, double value) { + if (!s->tail) { + s->head = s->tail = tsChunkCreate(s->chunk_size); + if (!s->tail) + return -1; + } + int rc = tsChunkAppend(s->tail, ts, value); + if (rc == 1) { + tsChunk *n = tsChunkCreate(s->chunk_size); + if (!n) + return -1; + s->tail->next = n; + s->tail = n; + rc = tsChunkAppend(s->tail, ts, value); + } + if (rc != 0) + return -1; + s->total_samples++; + s->last_ts = ts; + s->last_value = value; + s->has_sample = 1; + return 0; +} + +static int rebuild_from_samples(tsSeries *s, const tsSample *samples, + size_t n) { + series_clear_chunks(s); + for (size_t i = 0; i < n; i++) { + if (chunk_push_sample(s, samples[i].ts, samples[i].value) != 0) + return -1; + } + return 0; +} + +typedef struct gather_state { + tsSample *samples; + size_t n; + size_t cap; + int error; +} gather_state; + +static int gather_all(const tsSample *sample, void *ud) { + gather_state *st = ud; + if (st->n == st->cap) { + size_t ncap = st->cap ? st->cap * 2 : 64; + tsSample *grown = realloc(st->samples, sizeof(tsSample) * ncap); + if (!grown) { + st->error = 1; + return -1; + } + st->samples = grown; + st->cap = ncap; + } + st->samples[st->n++] = *sample; + return 0; +} + +static int load_all_samples(const tsSeries *s, tsSample **out, size_t *n) { + gather_state st; + memset(&st, 0, sizeof(st)); + if (tsSeriesVisit(s, gather_all, &st) != 0 || st.error) { + free(st.samples); + return -1; + } + *out = st.samples; + *n = st.n; + return 0; +} + +static int trim_retention(tsSeries *s) { + if (s->retention_ms <= 0 || !s->has_sample) + return 0; + int64_t cutoff = s->last_ts - s->retention_ms; + int dropped_whole = 0; + while (s->head && s->head->last_ts < cutoff) { + tsChunk *dead = s->head; + s->head = dead->next; + if (s->total_samples >= dead->count) + s->total_samples -= dead->count; + else + s->total_samples = 0; + tsChunkFree(dead); + dropped_whole = 1; + } + if (!s->head) { + s->tail = NULL; + s->has_sample = 0; + s->total_samples = 0; + return 0; + } + if (dropped_whole && s->head) { + /* tail pointer is unchanged unless we dropped everything. */ + } + if (s->head->first_ts < cutoff) { + tsSample *samples = NULL; + uint32_t n = 0; + if (tsChunkDecode(s->head, &samples, &n) != 0) + return -1; + uint32_t keep = 0; + for (uint32_t i = 0; i < n; i++) { + if (samples[i].ts >= cutoff) + samples[keep++] = samples[i]; + } + uint32_t removed = n - keep; + if (s->total_samples >= removed) + s->total_samples -= removed; + else + s->total_samples = 0; + if (tsChunkRecompress(s->head, samples, keep) != 0) { + free(samples); + return -1; + } + free(samples); + } + if (s->total_samples == 0) + s->has_sample = 0; + return 0; +} + +static int rewrite_last_sample(tsSeries *s, double value) { + if (!s->tail || s->tail->count == 0) + return -1; + tsSample *samples = NULL; + uint32_t n = 0; + if (tsChunkDecode(s->tail, &samples, &n) != 0 || n == 0) { + free(samples); + return -1; + } + samples[n - 1].value = value; + int rc = tsChunkRecompress(s->tail, samples, n); + free(samples); + if (rc != 0) + return -1; + s->last_value = value; + return 0; +} + +/* 0 stored, 1 duplicate ignored (FIRST), -1 block, -2 older, -3 oom. */ +static int series_add(tsSeries *s, int64_t ts, double value) { + if (s->has_sample && ts < s->last_ts) + return -2; + if (s->has_sample && ts == s->last_ts) { + double updated = value; + if (s->dup_policy == TS_DUP_BLOCK) + return -1; + if (s->dup_policy == TS_DUP_FIRST) + return 1; + if (s->dup_policy == TS_DUP_MIN) + updated = value < s->last_value ? value : s->last_value; + else if (s->dup_policy == TS_DUP_MAX) + updated = value > s->last_value ? value : s->last_value; + else if (s->dup_policy == TS_DUP_SUM) + updated = s->last_value + value; + if (rewrite_last_sample(s, updated) != 0) + return -3; + return 0; + } + if (chunk_push_sample(s, ts, value) != 0) + return -3; + if (trim_retention(s) != 0) + return -3; + return 0; +} + +static int series_delete_range(tsSeries *s, int64_t from, int64_t to, + long long *removed) { + tsSample *samples = NULL; + size_t n = 0; + if (load_all_samples(s, &samples, &n) != 0) + return -1; + size_t keep = 0; + long long gone = 0; + for (size_t i = 0; i < n; i++) { + if (samples[i].ts >= from && samples[i].ts <= to) { + gone++; + continue; + } + samples[keep++] = samples[i]; + } + if (rebuild_from_samples(s, samples, keep) != 0) { + free(samples); + return -1; + } + free(samples); + *removed = gone; + return 0; +} + +static int label_equals(const tsSeries *s, const char *key, const char *val) { + for (int i = 0; i < s->nlabels; i++) { + if (strcmp(s->labels[i].key, key) == 0 && + strcmp(s->labels[i].val, val) == 0) + return 1; + } + return 0; +} + +typedef struct tsFilter { + char *expr; + char *key; + char *val; + int negate; +} tsFilter; + +static int parse_filter(const char *expr, tsFilter *f) { + const char *neq = strstr(expr, "!="); + const char *eq = strchr(expr, '='); + f->expr = NULL; + f->key = NULL; + f->val = NULL; + f->negate = 0; + if (neq) { + f->negate = 1; + f->key = dup_cstr(expr, (size_t)(neq - expr)); + f->val = strdup(neq + 2); + } else if (eq && eq != expr) { + f->key = dup_cstr(expr, (size_t)(eq - expr)); + f->val = strdup(eq + 1); + } else { + return -1; + } + if (!f->key || !f->val || f->key[0] == '\0') { + free(f->key); + free(f->val); + return -1; + } + return 0; +} + +static int filters_match(const tsSeries *s, const tsFilter *filters, + int nfilters) { + for (int i = 0; i < nfilters; i++) { + int eq = label_equals(s, filters[i].key, filters[i].val); + if (filters[i].negate) { + if (eq) + return 0; + } else if (!eq) { + return 0; + } + } + return 1; +} + +static void reply_labels(ValkeyModuleCtx *ctx, const tsSeries *s) { + ValkeyModule_ReplyWithArray(ctx, s->nlabels); + for (int i = 0; i < s->nlabels; i++) { + ValkeyModule_ReplyWithArray(ctx, 2); + ValkeyModule_ReplyWithCString(ctx, s->labels[i].key); + ValkeyModule_ReplyWithCString(ctx, s->labels[i].val); + } +} + +static int parse_timestamp(ValkeyModuleCtx *ctx, ValkeyModuleString *arg, + int64_t *out, int allow_star) { + size_t len = 0; + const char *p = ValkeyModule_StringPtrLen(arg, &len); + if (allow_star && len == 1 && p[0] == '*') { + *out = (int64_t)ValkeyModule_Milliseconds(); + return 0; + } + if (len == 1 && p[0] == '-') { + *out = INT64_MIN; + return 0; + } + if (len == 1 && p[0] == '+') { + *out = INT64_MAX; + return 0; + } + long long v; + if (ValkeyModule_StringToLongLong(arg, &v) != VALKEYMODULE_OK || v < 0) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid timestamp"); + return -1; + } + *out = (int64_t)v; + return 0; +} + +static int open_series(ValkeyModuleCtx *ctx, ValkeyModuleString *keyname, + tsSeries **out, ValkeyModuleKey **kout, int create, + int64_t retention, int dup_policy, size_t chunk_size) { + ValkeyModuleKey *key = ValkeyModule_OpenKey( + ctx, keyname, VALKEYMODULE_READ | VALKEYMODULE_WRITE); + int type = ValkeyModule_KeyType(key); + if (type != VALKEYMODULE_KEYTYPE_EMPTY && + ValkeyModule_ModuleTypeGetType(key) != TsType) { + ValkeyModule_ReplyWithError(ctx, VALKEYMODULE_ERRORMSG_WRONGTYPE); + return -1; + } + tsSeries *s = NULL; + if (type == VALKEYMODULE_KEYTYPE_EMPTY) { + if (!create) { + *out = NULL; + *kout = key; + return 0; + } + s = series_create(chunk_size, retention, dup_policy); + if (!s) { + ValkeyModule_ReplyWithError(ctx, "ERR out of memory"); + return -1; + } + size_t klen = 0; + const char *k = ValkeyModule_StringPtrLen(keyname, &klen); + series_set_key(s, k, klen); + ValkeyModule_ModuleTypeSetValue(key, TsType, s); + } else { + s = ValkeyModule_ModuleTypeGetValue(key); + } + *out = s; + *kout = key; + return 0; +} + +static int parse_create_options(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc, int start, int64_t *retention, + int *dup_policy, size_t *chunk_size, + int *labels_at) { + *labels_at = -1; + for (int i = start; i < argc; i++) { + size_t len = 0; + const char *tok = ValkeyModule_StringPtrLen(argv[i], &len); + if (!strcasecmp(tok, "RETENTION")) { + long long v; + if (i + 1 >= argc || + ValkeyModule_StringToLongLong(argv[++i], &v) != VALKEYMODULE_OK || + v < 0) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid RETENTION"); + return -1; + } + *retention = (int64_t)v; + } else if (!strcasecmp(tok, "DUPLICATE_POLICY") || + !strcasecmp(tok, "ON_DUPLICATE")) { + if (i + 1 >= argc) { + ValkeyModule_ReplyWithError(ctx, "ERR missing duplicate policy"); + return -1; + } + size_t plen = 0; + const char *pol = ValkeyModule_StringPtrLen(argv[++i], &plen); + int parsed = tsDupPolicyFromName(pol); + if (parsed < 0) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid duplicate policy"); + return -1; + } + *dup_policy = parsed; + } else if (!strcasecmp(tok, "CHUNK_SIZE")) { + long long v; + if (i + 1 >= argc || + ValkeyModule_StringToLongLong(argv[++i], &v) != VALKEYMODULE_OK || + v <= 0) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid CHUNK_SIZE"); + return -1; + } + *chunk_size = (size_t)v; + } else if (!strcasecmp(tok, "ENCODING")) { + if (i + 1 >= argc) { + ValkeyModule_ReplyWithError(ctx, "ERR missing ENCODING"); + return -1; + } + i++; /* Accepted for clients. Samples are always stored compressed. */ + } else if (!strcasecmp(tok, "LABELS")) { + *labels_at = i + 1; + if (((argc - *labels_at) % 2) != 0) { + ValkeyModule_ReplyWithError(ctx, "ERR LABELS requires key value pairs"); + return -1; + } + return 0; + } else { + ValkeyModule_ReplyWithError(ctx, "ERR syntax error"); + return -1; + } + } + return 0; +} + +static int apply_labels(ValkeyModuleCtx *ctx, tsSeries *s, + ValkeyModuleString **argv, int begin, int end) { + for (int i = begin; i + 1 < end; i += 2) { + size_t klen = 0, vlen = 0; + const char *k = ValkeyModule_StringPtrLen(argv[i], &klen); + const char *v = ValkeyModule_StringPtrLen(argv[i + 1], &vlen); + char *ks = dup_cstr(k, klen); + char *vs = dup_cstr(v, vlen); + if (!ks || !vs || series_set_label(s, ks, vs) != 0) { + free(ks); + free(vs); + ValkeyModule_ReplyWithError(ctx, "ERR out of memory"); + return -1; + } + free(ks); + free(vs); + } + return 0; +} + +static int TsCreate_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + ValkeyModule_AutoMemory(ctx); + if (argc < 2) + return ValkeyModule_WrongArity(ctx); + int64_t retention = 0; + int dup_policy = TS_DUP_BLOCK; + size_t chunk_size = TS_CHUNK_DEFAULT_BYTES; + int labels_at = -1; + if (parse_create_options(ctx, argv, argc, 2, &retention, &dup_policy, + &chunk_size, &labels_at) != 0) + return VALKEYMODULE_OK; + + ValkeyModuleKey *key = ValkeyModule_OpenKey( + ctx, argv[1], VALKEYMODULE_READ | VALKEYMODULE_WRITE); + if (ValkeyModule_KeyType(key) != VALKEYMODULE_KEYTYPE_EMPTY) { + return ValkeyModule_ReplyWithError(ctx, "ERR key already exists"); + } + tsSeries *s = series_create(chunk_size, retention, dup_policy); + if (!s) + return ValkeyModule_ReplyWithError(ctx, "ERR out of memory"); + size_t klen = 0; + const char *k = ValkeyModule_StringPtrLen(argv[1], &klen); + series_set_key(s, k, klen); + if (labels_at >= 0 && apply_labels(ctx, s, argv, labels_at, argc) != 0) { + series_free(s); + return VALKEYMODULE_OK; + } + ValkeyModule_ModuleTypeSetValue(key, TsType, s); + ValkeyModule_ReplicateVerbatim(ctx); + return ValkeyModule_ReplyWithSimpleString(ctx, "OK"); +} + +static int add_one(ValkeyModuleCtx *ctx, ValkeyModuleString *keyname, + int64_t ts, double value, int create, int64_t retention, + int dup_policy, size_t chunk_size, int64_t *stored_ts) { + tsSeries *s = NULL; + ValkeyModuleKey *key = NULL; + if (open_series(ctx, keyname, &s, &key, create, retention, dup_policy, + chunk_size) != 0) + return -1; + if (!s) { + ValkeyModule_ReplyWithError(ctx, "ERR key does not exist"); + return -1; + } + /* Creation options on TS.ADD apply only when this command created the key. + * An existing series keeps the policy it was created with. */ + int rc = series_add(s, ts, value); + if (rc == -1) { + ValkeyModule_ReplyWithError(ctx, "ERR duplicate sample"); + return -1; + } + if (rc == -2) { + ValkeyModule_ReplyWithError( + ctx, "ERR timestamp is older than the latest sample"); + return -1; + } + if (rc == -3) { + ValkeyModule_ReplyWithError(ctx, "ERR out of memory"); + return -1; + } + *stored_ts = ts; + return 0; +} + +static int TsAdd_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + ValkeyModule_AutoMemory(ctx); + if (argc < 4) + return ValkeyModule_WrongArity(ctx); + int64_t ts; + if (parse_timestamp(ctx, argv[2], &ts, 1) != 0) + return VALKEYMODULE_OK; + if (ts < 0) + return ValkeyModule_ReplyWithError(ctx, "ERR invalid timestamp"); + double value; + if (ValkeyModule_StringToDouble(argv[3], &value) != VALKEYMODULE_OK) { + return ValkeyModule_ReplyWithError(ctx, "ERR invalid value"); + } + int64_t retention = 0; + int dup_policy = TS_DUP_BLOCK; + size_t chunk_size = TS_CHUNK_DEFAULT_BYTES; + int labels_at = -1; + if (argc > 4 && + parse_create_options(ctx, argv, argc, 4, &retention, &dup_policy, + &chunk_size, &labels_at) != 0) + return VALKEYMODULE_OK; + + int64_t stored = 0; + if (add_one(ctx, argv[1], ts, value, 1, retention, dup_policy, chunk_size, + &stored) != 0) + return VALKEYMODULE_OK; + tsSeries *s = NULL; + ValkeyModuleKey *key = ValkeyModule_OpenKey( + ctx, argv[1], VALKEYMODULE_READ | VALKEYMODULE_WRITE); + s = ValkeyModule_ModuleTypeGetValue(key); + if (s && labels_at >= 0 && s->total_samples <= 1 && + apply_labels(ctx, s, argv, labels_at, argc) != 0) + return VALKEYMODULE_OK; + ValkeyModule_ReplicateVerbatim(ctx); + return ValkeyModule_ReplyWithLongLong(ctx, (long long)stored); +} + +static int TsMadd_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + ValkeyModule_AutoMemory(ctx); + if (argc < 4 || ((argc - 1) % 3) != 0) + return ValkeyModule_WrongArity(ctx); + int n = (argc - 1) / 3; + ValkeyModule_ReplyWithArray(ctx, n); + for (int i = 0; i < n; i++) { + int base = 1 + i * 3; + int64_t ts; + size_t len = 0; + const char *p = ValkeyModule_StringPtrLen(argv[base + 1], &len); + int ts_ok = 1; + if (len == 1 && p[0] == '*') + ts = (int64_t)ValkeyModule_Milliseconds(); + else { + long long v; + if (ValkeyModule_StringToLongLong(argv[base + 1], &v) != + VALKEYMODULE_OK || + v < 0) + ts_ok = 0; + else + ts = (int64_t)v; + } + double value; + int val_ok = + ValkeyModule_StringToDouble(argv[base + 2], &value) == VALKEYMODULE_OK; + if (!ts_ok || !val_ok) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid timestamp or value"); + continue; + } + /* ReplyWithError inside an array element is not how the module API + * reports a per-item error. Use a bulk error string instead. */ + ValkeyModuleKey *key = ValkeyModule_OpenKey( + ctx, argv[base], VALKEYMODULE_READ | VALKEYMODULE_WRITE); + int type = ValkeyModule_KeyType(key); + if (type != VALKEYMODULE_KEYTYPE_EMPTY && + ValkeyModule_ModuleTypeGetType(key) != TsType) { + ValkeyModule_ReplyWithSimpleString(ctx, "ERR wrong type"); + continue; + } + tsSeries *s = NULL; + if (type == VALKEYMODULE_KEYTYPE_EMPTY) { + s = series_create(TS_CHUNK_DEFAULT_BYTES, 0, TS_DUP_BLOCK); + if (!s) { + ValkeyModule_ReplyWithSimpleString(ctx, "ERR out of memory"); + continue; + } + size_t klen = 0; + const char *k = ValkeyModule_StringPtrLen(argv[base], &klen); + series_set_key(s, k, klen); + ValkeyModule_ModuleTypeSetValue(key, TsType, s); + } else { + s = ValkeyModule_ModuleTypeGetValue(key); + } + int rc = series_add(s, ts, value); + if (rc < 0) { + const char *msg = rc == -2 + ? "ERR timestamp is older than the latest sample" + : rc == -1 ? "ERR duplicate sample" + : "ERR out of memory"; + ValkeyModule_ReplyWithSimpleString(ctx, msg); + continue; + } + ValkeyModule_ReplyWithLongLong(ctx, (long long)ts); + } + ValkeyModule_ReplicateVerbatim(ctx); + return VALKEYMODULE_OK; +} + +static int TsGet_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + ValkeyModule_AutoMemory(ctx); + if (argc != 2) + return ValkeyModule_WrongArity(ctx); + tsSeries *s = NULL; + ValkeyModuleKey *key = NULL; + if (open_series(ctx, argv[1], &s, &key, 0, 0, TS_DUP_BLOCK, + TS_CHUNK_DEFAULT_BYTES) != 0) + return VALKEYMODULE_OK; + if (!s || !s->has_sample) + return ValkeyModule_ReplyWithArray(ctx, 0); + ValkeyModule_ReplyWithArray(ctx, 2); + ValkeyModule_ReplyWithLongLong(ctx, (long long)s->last_ts); + ValkeyModule_ReplyWithDouble(ctx, s->last_value); + return VALKEYMODULE_OK; +} + +static int parse_range_suffix(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc, int start, int *agg, int64_t *bucket, + long long *count, int *filter_value, double *vmin, + double *vmax) { + *agg = TS_AGG_NONE; + *bucket = 0; + *count = -1; + *filter_value = 0; + for (int i = start; i < argc; i++) { + const char *tok = ValkeyModule_StringPtrLen(argv[i], NULL); + if (!strcasecmp(tok, "AGGREGATION")) { + if (i + 2 >= argc) { + ValkeyModule_ReplyWithError(ctx, "ERR syntax error"); + return -1; + } + *agg = tsAggFromName(ValkeyModule_StringPtrLen(argv[++i], NULL)); + long long b; + if (*agg < 0 || + ValkeyModule_StringToLongLong(argv[++i], &b) != VALKEYMODULE_OK || + b <= 0) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid AGGREGATION"); + return -1; + } + *bucket = (int64_t)b; + } else if (!strcasecmp(tok, "COUNT")) { + long long c; + if (i + 1 >= argc || + ValkeyModule_StringToLongLong(argv[++i], &c) != VALKEYMODULE_OK || + c < 0) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid COUNT"); + return -1; + } + *count = c; + } else if (!strcasecmp(tok, "FILTER_BY_VALUE")) { + if (i + 2 >= argc || + ValkeyModule_StringToDouble(argv[i + 1], vmin) != VALKEYMODULE_OK || + ValkeyModule_StringToDouble(argv[i + 2], vmax) != VALKEYMODULE_OK) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid FILTER_BY_VALUE"); + return -1; + } + i += 2; + *filter_value = 1; + } else { + ValkeyModule_ReplyWithError(ctx, "ERR syntax error"); + return -1; + } + } + return 0; +} + +static int range_command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc, int reverse) { + ValkeyModule_AutoMemory(ctx); + if (argc < 4) + return ValkeyModule_WrongArity(ctx); + int64_t from, to; + if (parse_timestamp(ctx, argv[2], &from, 0) != 0) + return VALKEYMODULE_OK; + if (parse_timestamp(ctx, argv[3], &to, 0) != 0) + return VALKEYMODULE_OK; + if (from == INT64_MIN) + from = 0; + int agg, filter_value; + int64_t bucket; + long long count; + double vmin, vmax; + if (parse_range_suffix(ctx, argv, argc, 4, &agg, &bucket, &count, + &filter_value, &vmin, &vmax) != 0) + return VALKEYMODULE_OK; + tsSeries *s = NULL; + ValkeyModuleKey *key = NULL; + if (open_series(ctx, argv[1], &s, &key, 0, 0, TS_DUP_BLOCK, + TS_CHUNK_DEFAULT_BYTES) != 0) + return VALKEYMODULE_OK; + if (tsReplySamples(ctx, s, from, to, reverse, agg, bucket, count, + filter_value, vmin, vmax) != 0) { + return ValkeyModule_ReplyWithError(ctx, "ERR query failed"); + } + return VALKEYMODULE_OK; +} + +static int TsRange_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + return range_command(ctx, argv, argc, 0); +} + +static int TsRevRange_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + return range_command(ctx, argv, argc, 1); +} + +typedef struct mrange_hit { + tsSeries *series; + char *key; +} mrange_hit; + +typedef struct mrange_scan { + tsFilter *filters; + int nfilters; + mrange_hit *hits; + int n; + int cap; +} mrange_scan; + +/* Called by ValkeyModule_Scan. The key object is owned by the scan when it + * is passed in. A missing key is opened for this call only; AutoMemory on + * the command context releases that handle. */ +static void mrange_on_key(ValkeyModuleCtx *ctx, ValkeyModuleString *keyname, + ValkeyModuleKey *key, void *privdata) { + mrange_scan *st = privdata; + if (!key) + key = ValkeyModule_OpenKey(ctx, keyname, VALKEYMODULE_READ); + if (!key) + return; + if (ValkeyModule_KeyType(key) != VALKEYMODULE_KEYTYPE_MODULE || + ValkeyModule_ModuleTypeGetType(key) != TsType) + return; + tsSeries *s = ValkeyModule_ModuleTypeGetValue(key); + if (!s || !filters_match(s, st->filters, st->nfilters)) + return; + if (st->n == st->cap) { + int ncap = st->cap ? st->cap * 2 : 8; + mrange_hit *grown = realloc(st->hits, sizeof(*grown) * (size_t)ncap); + if (!grown) + return; + st->hits = grown; + st->cap = ncap; + } + size_t klen = 0; + const char *k = ValkeyModule_StringPtrLen(keyname, &klen); + st->hits[st->n].series = s; + st->hits[st->n].key = dup_cstr(k, klen); + st->n++; +} + +static int TsMrange_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + ValkeyModule_AutoMemory(ctx); + if (argc < 4) + return ValkeyModule_WrongArity(ctx); + int64_t from, to; + if (parse_timestamp(ctx, argv[1], &from, 0) != 0) + return VALKEYMODULE_OK; + if (parse_timestamp(ctx, argv[2], &to, 0) != 0) + return VALKEYMODULE_OK; + if (from < 0) + from = 0; + int withlabels = 0; + int filter_at = -1; + int agg = TS_AGG_NONE; + int64_t bucket = 0; + long long count = -1; + for (int i = 3; i < argc; i++) { + const char *tok = ValkeyModule_StringPtrLen(argv[i], NULL); + if (!strcasecmp(tok, "WITHLABELS")) { + withlabels = 1; + } else if (!strcasecmp(tok, "FILTER")) { + filter_at = i + 1; + break; + } else if (!strcasecmp(tok, "AGGREGATION")) { + if (i + 2 >= argc) + return ValkeyModule_ReplyWithError(ctx, "ERR syntax error"); + agg = tsAggFromName(ValkeyModule_StringPtrLen(argv[++i], NULL)); + long long b; + if (agg < 0 || + ValkeyModule_StringToLongLong(argv[++i], &b) != VALKEYMODULE_OK || + b <= 0) { + return ValkeyModule_ReplyWithError(ctx, "ERR invalid AGGREGATION"); + } + bucket = (int64_t)b; + } else if (!strcasecmp(tok, "COUNT")) { + long long c; + if (i + 1 >= argc || + ValkeyModule_StringToLongLong(argv[++i], &c) != VALKEYMODULE_OK || + c < 0) { + return ValkeyModule_ReplyWithError(ctx, "ERR invalid COUNT"); + } + count = c; + } else { + return ValkeyModule_ReplyWithError(ctx, "ERR syntax error"); + } + } + if (filter_at < 0 || filter_at >= argc) + return ValkeyModule_ReplyWithError(ctx, "ERR FILTER is required"); + int nfilters = argc - filter_at; + tsFilter *filters = calloc((size_t)nfilters, sizeof(tsFilter)); + if (!filters) + return ValkeyModule_ReplyWithError(ctx, "ERR out of memory"); + for (int i = 0; i < nfilters; i++) { + if (parse_filter(ValkeyModule_StringPtrLen(argv[filter_at + i], NULL), + &filters[i]) != 0) { + ValkeyModule_ReplyWithError(ctx, "ERR invalid filter"); + for (int j = 0; j < i; j++) { + free(filters[j].key); + free(filters[j].val); + } + free(filters); + return VALKEYMODULE_OK; + } + } + + /* Scan the live keyspace. A side index would still point at values that + * FLUSH / DEBUG RELOAD has unlinked but not freed yet (lazy free runs off + * the main thread). */ + mrange_scan scan; + memset(&scan, 0, sizeof(scan)); + scan.filters = filters; + scan.nfilters = nfilters; + ValkeyModuleScanCursor *cursor = ValkeyModule_ScanCursorCreate(); + while (ValkeyModule_Scan(ctx, cursor, mrange_on_key, &scan)) { + } + ValkeyModule_ScanCursorDestroy(cursor); + + ValkeyModule_ReplyWithArray(ctx, scan.n); + for (int i = 0; i < scan.n; i++) { + tsSeries *s = scan.hits[i].series; + ValkeyModule_ReplyWithArray(ctx, 3); + ValkeyModule_ReplyWithCString(ctx, + scan.hits[i].key ? scan.hits[i].key : ""); + if (withlabels) + reply_labels(ctx, s); + else + ValkeyModule_ReplyWithArray(ctx, 0); + if (tsReplySamples(ctx, s, from, to, 0, agg, bucket, count, 0, 0, 0) != 0) { + ValkeyModule_ReplyWithArray(ctx, 0); + } + free(scan.hits[i].key); + } + free(scan.hits); + for (int i = 0; i < nfilters; i++) { + free(filters[i].key); + free(filters[i].val); + } + free(filters); + return VALKEYMODULE_OK; +} + +static int TsDel_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + ValkeyModule_AutoMemory(ctx); + if (argc != 4) + return ValkeyModule_WrongArity(ctx); + int64_t from, to; + if (parse_timestamp(ctx, argv[2], &from, 0) != 0) + return VALKEYMODULE_OK; + if (parse_timestamp(ctx, argv[3], &to, 0) != 0) + return VALKEYMODULE_OK; + if (from < 0) + from = 0; + tsSeries *s = NULL; + ValkeyModuleKey *key = NULL; + if (open_series(ctx, argv[1], &s, &key, 0, 0, TS_DUP_BLOCK, + TS_CHUNK_DEFAULT_BYTES) != 0) + return VALKEYMODULE_OK; + if (!s) + return ValkeyModule_ReplyWithLongLong(ctx, 0); + long long removed = 0; + if (series_delete_range(s, from, to, &removed) != 0) + return ValkeyModule_ReplyWithError(ctx, "ERR delete failed"); + ValkeyModule_ReplicateVerbatim(ctx); + return ValkeyModule_ReplyWithLongLong(ctx, removed); +} + +static int TsLabel_Command(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + ValkeyModule_AutoMemory(ctx); + if (argc != 4) + return ValkeyModule_WrongArity(ctx); + tsSeries *s = NULL; + ValkeyModuleKey *key = NULL; + if (open_series(ctx, argv[1], &s, &key, 0, 0, TS_DUP_BLOCK, + TS_CHUNK_DEFAULT_BYTES) != 0) + return VALKEYMODULE_OK; + if (!s) + return ValkeyModule_ReplyWithError(ctx, "ERR key does not exist"); + if (apply_labels(ctx, s, argv, 2, 4) != 0) + return VALKEYMODULE_OK; + ValkeyModule_ReplicateVerbatim(ctx); + return ValkeyModule_ReplyWithSimpleString(ctx, "OK"); +} + +static void *TsRdbLoad(ValkeyModuleIO *rdb, int encver) { + if (encver != 1) + return NULL; + int64_t retention = (int64_t)ValkeyModule_LoadSigned(rdb); + int dup_policy = (int)ValkeyModule_LoadUnsigned(rdb); + uint64_t chunk_size = ValkeyModule_LoadUnsigned(rdb); + uint64_t nlabels = ValkeyModule_LoadUnsigned(rdb); + tsSeries *s = series_create((size_t)chunk_size, retention, dup_policy); + if (!s) + return NULL; + for (uint64_t i = 0; i < nlabels; i++) { + size_t klen = 0, vlen = 0; + char *k = ValkeyModule_LoadStringBuffer(rdb, &klen); + char *v = ValkeyModule_LoadStringBuffer(rdb, &vlen); + if (!k || !v) { + if (k) + ValkeyModule_Free(k); + if (v) + ValkeyModule_Free(v); + series_free(s); + return NULL; + } + char *ks = dup_cstr(k, klen); + char *vs = dup_cstr(v, vlen); + ValkeyModule_Free(k); + ValkeyModule_Free(v); + if (!ks || !vs || series_set_label(s, ks, vs) != 0) { + free(ks); + free(vs); + series_free(s); + return NULL; + } + free(ks); + free(vs); + } + uint64_t nsamples = ValkeyModule_LoadUnsigned(rdb); + for (uint64_t i = 0; i < nsamples; i++) { + int64_t ts = (int64_t)ValkeyModule_LoadSigned(rdb); + double value = ValkeyModule_LoadDouble(rdb); + if (chunk_push_sample(s, ts, value) != 0) { + series_free(s); + return NULL; + } + } + const ValkeyModuleString *keyname = ValkeyModule_GetKeyNameFromIO(rdb); + if (keyname) { + size_t klen = 0; + const char *k = ValkeyModule_StringPtrLen(keyname, &klen); + series_set_key(s, k, klen); + } + if (ValkeyModule_IsIOError(rdb)) { + series_free(s); + return NULL; + } + return s; +} + +static void TsRdbSave(ValkeyModuleIO *rdb, void *value) { + tsSeries *s = value; + ValkeyModule_SaveSigned(rdb, s->retention_ms); + ValkeyModule_SaveUnsigned(rdb, (uint64_t)s->dup_policy); + ValkeyModule_SaveUnsigned(rdb, (uint64_t)s->chunk_size); + ValkeyModule_SaveUnsigned(rdb, (uint64_t)s->nlabels); + for (int i = 0; i < s->nlabels; i++) { + ValkeyModule_SaveStringBuffer(rdb, s->labels[i].key, + strlen(s->labels[i].key)); + ValkeyModule_SaveStringBuffer(rdb, s->labels[i].val, + strlen(s->labels[i].val)); + } + tsSample *samples = NULL; + size_t n = 0; + if (load_all_samples(s, &samples, &n) != 0) { + ValkeyModule_SaveUnsigned(rdb, 0); + return; + } + ValkeyModule_SaveUnsigned(rdb, (uint64_t)n); + for (size_t i = 0; i < n; i++) { + ValkeyModule_SaveSigned(rdb, samples[i].ts); + ValkeyModule_SaveDouble(rdb, samples[i].value); + } + free(samples); +} + +static void TsAofRewrite(ValkeyModuleIO *aof, ValkeyModuleString *key, + void *value) { + tsSeries *s = value; + ValkeyModule_EmitAOF(aof, "TS.CREATE", "sclcc", key, "RETENTION", + (long long)s->retention_ms, "DUPLICATE_POLICY", + tsDupPolicyName(s->dup_policy)); + for (int i = 0; i < s->nlabels; i++) { + ValkeyModule_EmitAOF(aof, "TS.LABEL", "scc", key, s->labels[i].key, + s->labels[i].val); + } + tsSample *samples = NULL; + size_t n = 0; + if (load_all_samples(s, &samples, &n) != 0) + return; + for (size_t i = 0; i < n; i++) { + char buf[64]; + snprintf(buf, sizeof(buf), "%.17g", samples[i].value); + ValkeyModule_EmitAOF(aof, "TS.ADD", "slc", key, (long long)samples[i].ts, + buf); + } + free(samples); +} + +static size_t TsMemUsage(const void *value) { + return tsSeriesStorageBytes(value); +} + +static void TsFree(void *value) { series_free(value); } + +static void TsDigest(ValkeyModuleDigest *md, void *value) { + tsSeries *s = value; + tsSample *samples = NULL; + size_t n = 0; + if (load_all_samples(s, &samples, &n) != 0) + return; + for (size_t i = 0; i < n; i++) { + ValkeyModule_DigestAddLongLong(md, samples[i].ts); + uint64_t bits; + memcpy(&bits, &samples[i].value, sizeof(bits)); + ValkeyModule_DigestAddStringBuffer(md, (char *)&bits, sizeof(bits)); + } + free(samples); + ValkeyModule_DigestEndSequence(md); +} + +int ValkeyModule_OnLoad(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, + int argc) { + VALKEYMODULE_NOT_USED(argv); + VALKEYMODULE_NOT_USED(argc); + if (ValkeyModule_Init(ctx, "timeseries", 1, VALKEYMODULE_APIVER_1) == + VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + + ValkeyModuleTypeMethods tm = {.version = VALKEYMODULE_TYPE_METHOD_VERSION, + .rdb_load = TsRdbLoad, + .rdb_save = TsRdbSave, + .aof_rewrite = TsAofRewrite, + .mem_usage = TsMemUsage, + .free = TsFree, + .digest = TsDigest}; + TsType = ValkeyModule_CreateDataType(ctx, "cmo-ts-v1", 1, &tm); + if (TsType == NULL) + return VALKEYMODULE_ERR; + + if (ValkeyModule_CreateCommand(ctx, "ts.create", TsCreate_Command, + "write deny-oom", 1, 1, 1) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.add", TsAdd_Command, "write deny-oom", + 1, 1, 1) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.madd", TsMadd_Command, + "write deny-oom", 1, -3, + 3) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.get", TsGet_Command, "readonly", 1, 1, + 1) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.range", TsRange_Command, "readonly", + 1, 1, 1) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.revrange", TsRevRange_Command, + "readonly", 1, 1, 1) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.mrange", TsMrange_Command, "readonly", + 0, 0, 0) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.del", TsDel_Command, "write", 1, 1, + 1) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + if (ValkeyModule_CreateCommand(ctx, "ts.label", TsLabel_Command, + "write deny-oom", 1, 1, 1) == VALKEYMODULE_ERR) + return VALKEYMODULE_ERR; + return VALKEYMODULE_OK; +} diff --git a/tests/modules/timeseries.tcl b/tests/modules/timeseries.tcl new file mode 100644 index 00000000000..130402ff619 --- /dev/null +++ b/tests/modules/timeseries.tcl @@ -0,0 +1,141 @@ +# Time-series module. Compression fuzz lives in modules/timeseries/chunk_test. +# This file is sourced from tests/unit/timeseries.tcl so `make test` runs it. + +set ::ts_root [file normalize [file join [file dirname [info script]] .. ..]] +exec make -C [file join $::ts_root modules/timeseries] +set ::ts_module [file normalize [file join $::ts_root modules/timeseries/timeseries.so]] +set ::ts_chunk_test [file normalize [file join $::ts_root modules/timeseries/chunk_test]] + +test {timeseries compression fuzz, aggregation reference, and 1M budget} { + exec $::ts_chunk_test +} {*OK} + +start_server [list overrides [list loadmodule $::ts_module appendonly yes aof-use-rdb-preamble no]] { + test {TS.CREATE TS.ADD TS.GET} { + r ts.create temp RETENTION 0 DUPLICATE_POLICY block LABELS sensor a + r ts.add temp 10 1 + r ts.add temp 20 2 + r ts.get temp + } {20 2} + + test {duplicate policy BLOCK rejects the same timestamp} { + assert_error {*duplicate*} {r ts.add temp 20 9} + } + + test {duplicate policy LAST overwrites the sample} { + r ts.create lastp DUPLICATE_POLICY last + r ts.add lastp 1 1 + r ts.add lastp 1 5 + r ts.get lastp + } {1 5} + + test {duplicate policy MIN keeps the smaller value} { + r ts.create minp DUPLICATE_POLICY min + r ts.add minp 1 4 + r ts.add minp 1 2 + r ts.add minp 1 9 + set pair [r ts.get minp] + assert_equal [lindex $pair 0] 1 + assert {[expr {abs([lindex $pair 1] - 2)}] < 1e-9} + } + + test {TS.MADD and TS.RANGE} { + r ts.create walk CHUNK_SIZE 128 + r ts.madd walk 0 1 walk 10 3 walk 20 5 + set samples [r ts.range walk - +] + assert_equal [llength $samples] 3 + assert_equal [lindex [lindex $samples 0] 0] 0 + assert_equal [lindex [lindex $samples 2] 0] 20 + } + + test {TS.RANGE aggregation matches a naive bucket average} { + # Buckets of 100: [0,10,20] -> avg 3, [100] -> 10. + r ts.add walk 100 10 + set got [r ts.range walk - + AGGREGATION avg 100] + assert_equal [llength $got] 2 + assert_equal [lindex [lindex $got 0] 0] 0 + assert {[expr {abs([lindex [lindex $got 0] 1] - 3.0)}] < 1e-9} + assert_equal [lindex [lindex $got 1] 0] 100 + assert {[expr {abs([lindex [lindex $got 1] 1] - 10.0)}] < 1e-9} + set rev [r ts.revrange walk - + AGGREGATION sum 100] + assert_equal [lindex [lindex $rev 0] 0] 100 + assert {[expr {abs([lindex [lindex $rev 0] 1] - 10.0)}] < 1e-9} + assert {[expr {abs([lindex [lindex $rev 1] 1] - 9.0)}] < 1e-9} + } + + test {TS.RANGE COUNT and FILTER_BY_VALUE} { + set counted [r ts.range walk 0 20 COUNT 2] + assert_equal [llength $counted] 2 + set filtered [r ts.range walk - + FILTER_BY_VALUE 3 5] + assert_equal [llength $filtered] 2 + } + + test {TS.MRANGE label filters} { + r ts.create kitchen LABELS room kitchen color red + r ts.create bath LABELS room bath color blue + r ts.add kitchen 1 7 + r ts.add bath 1 8 + set hit [r ts.mrange - + WITHLABELS FILTER room=kitchen] + assert_equal [llength $hit] 1 + assert_equal [lindex [lindex $hit 0] 0] kitchen + set labels [lindex [lindex $hit 0] 1] + assert_equal [llength $labels] 2 + set samples [lindex [lindex $hit 0] 2] + assert_equal [lindex [lindex $samples 0] 0] 1 + set miss [r ts.mrange - + FILTER room=kitchen color!=red] + assert_equal [llength $miss] 0 + set either [r ts.mrange - + FILTER room!=garage] + assert {[llength $either] >= 2} + } + + test {TS.DEL removes a closed timestamp range} { + set removed [r ts.del walk 0 10] + assert_equal $removed 2 + set left [r ts.range walk - +] + assert_equal [llength $left] 2 + assert_equal [lindex [lindex $left 0] 0] 20 + } + + test {retention drops samples older than the newest minus the window} { + r ts.create aged RETENTION 1000 + r ts.add aged 1000 1 + r ts.add aged 5000 2 + set got [r ts.range aged - +] + assert_equal [llength $got] 1 + assert_equal [lindex [lindex $got 0] 0] 5000 + } + + test {older timestamps are rejected} { + assert_error {*older*} {r ts.add aged 1000 1} + } + + test {RDB save and reload keeps samples and labels} { + r save + r debug reload + set got [r ts.range walk - +] + assert_equal [llength $got] 2 + assert_equal [lindex [lindex $got 0] 0] 20 + set hit [r ts.mrange - + FILTER room=kitchen] + assert_equal [llength $hit] 1 + assert_equal [lindex [lindex [lindex [lindex $hit 0] 2] 0] 0] 1 + } {} {needs:debug} + + test {AOF rewrite replays the series} { + r bgrewriteaof + waitForBgrewriteaof r + r debug loadaof + set got [r ts.get lastp] + assert_equal [lindex $got 0] 1 + assert {[expr {abs([lindex $got 1] - 5)}] < 1e-9} + set aged [r ts.range aged - +] + assert_equal [llength $aged] 1 + set hit [r ts.mrange - + WITHLABELS FILTER room=bath] + assert_equal [llength $hit] 1 + } {} {needs:debug} + + test {a string key is the wrong type} { + r set plain hello + assert_error {*WRONGTYPE*} {r ts.add plain 1 1} + assert_error {*WRONGTYPE*} {r ts.get plain} + } +} diff --git a/tests/unit/timeseries.tcl b/tests/unit/timeseries.tcl new file mode 100644 index 00000000000..a569a95e433 --- /dev/null +++ b/tests/unit/timeseries.tcl @@ -0,0 +1,3 @@ +# Discovered by the server test runner. The case itself lives next to the +# other module tests. +source [file join [file dirname [info script]] .. modules timeseries.tcl]