cpp-timeseries-downsample-dispatch

Answer

Inventory — what exists vs. what's missing (9 verified facts):

4 design guidance items given to the worker:

1. Storage — extend time_series_catalog_t with per-step B-tree roots, sindex-style (root block_id_t + counts in the superblock time_series_block_t), keyed by (step_name, target_interval_us). Bucket key = epoch-aligned interval start (int64 µs, floor division).
2. Merge — run_downsample_pass → store_t::run_downsample_step: consume only sealed chunks in the step's source range, bucket points by target interval, evaluate wire_func_t per bucket (two-phase bucket: open accumulates samples, sealed stores final value); durable per-chunk cursor for idempotent resume; periodic superblock checkpoint.
3. Planner — add a downsample spec to rget_read_t; in do_read's rget branch, select the step root and read only buckets intersecting ts_range (reuse the existing pruning on bucket keys); lower table.downsample(...) in ops.hpp into that rget.
4. Semantics — target interval must be a multiple of source (enforced at config); empty/non-numeric buckets → null; config change mid-pass obsoletes the running job (abort cleanly, durable cursor prevents partial commits).

The fix — code examples:

// ---- rdb_protocol/jobs.cc : replace the no-op run_downsample_pass ----
void run_downsample_pass(job_t *job, job_manager_t *job_manager) {
    const downsample_job_data_t &data = boost::get<downsample_job_data_t>(job->data);
    job->state = job_state_t::RUNNING;

    auto table = get_table(data.table_id);
    // Obsolete job: table dropped or step removed -> complete without work.
    if (!table.has() || !table->config().has_downsample_step(data.step_name)) {
        job_manager->complete(job);
        return;
    }
    // Each shard owns its sealed chunks + downsample roots; one pass per shard.
    for (auto &shard : table->shards()) {
        shard.store()->run_downsample_step(data.step_name, job_manager);
        if (job->cancelled()) break;   // config change / shutdown
    }
    job_manager->complete(job);
}

// ---- time_series_catalog.hpp : sindex-style per-step roots ----
struct downsample_root_t {
    std::string step_name;
    int64_t target_interval_us;        // bucket width in µs
    block_id_t root;                   // B-tree root over bucket-start keys
    uint64_t sealed_bucket_count;
    uint64_t point_count;
    uint64_t processed_chunk_cursor;   // durable resume point

    void rdb_serialize(write_message_t *wm) const;    // superblock wire format
    archive_result_t rdb_deserialize(read_stream_t *s);
};

class time_series_catalog_t {
public:
    // ... existing sealed/open chunk management unchanged ...
    void set_downsample_root(const downsample_root_t &r) {
        downsample_roots_[{r.step_name, r.target_interval_us}] = r;
    }
    const downsample_root_t *get_downsample_root(const std::string &name,
                                                 int64_t iv) const {
        auto it = downsample_roots_.find({name, iv});
        return it == downsample_roots_.end() ? nullptr : &it->second;
    }
    void erase_downsample_root(const std::string &name, int64_t iv);

    // Called from store_t's superblock persist: existing chunk catalog,
    // then the downsample roots vector (serialized last, like real_sindexes).
    void serialize_into(write_message_t *wm) const { /* chunk catalog ... */ serialize_vector(wm, root_vec_); }
private:
    std::map<std::pair<std::string, int64_t>, downsample_root_t> downsample_roots_;
};

// ---- store.cc : the merge pass ----
static int64_t bucket_of(int64_t ts_us, int64_t target_us) {
    int64_t q = ts_us / target_us, r = ts_us % target_us;
    if (r < 0) { q -= 1; }              // floor division: negatives align correctly
    return q * target_us;
}

void store_t::run_downsample_step(const std::string &step_name, job_manager_t *mgr) {
    const downsample_step_t &step = table_config().downsample_step(step_name);
    const int64_t source_us = step.source_interval.to_us();
    const int64_t target_us = step.target_interval.to_us();
    rassert(target_us % source_us == 0);               // enforced at config time

    time_series_catalog_t &cat = catalog();
    block_id_t root = get_or_allocate_downsample_root(step_name, target_us);

    uint64_t cursor = cat.get_downsample_root(step_name, target_us)->processed_chunk_cursor;
    std::vector<chunk_id_t> sealed = cat.sealed_chunks_in_range(step.source_range);

    for (; cursor < sealed.size(); ++cursor) {
        if (mgr->should_abort()) break;

        std::vector<point_t> points;
        read_sealed_chunk_points(sealed[cursor], &points);   // immutable chunk

        // Bucket the chunk's points (a chunk can feed several target buckets).
        std::map<int64_t, bucket_accum_t> buckets;
        for (const point_t &p : points)
            buckets[bucket_of(p.timestamp_us, target_us)].samples.push_back(p.value);

        // Seal any bucket whose interval is fully covered once the next chunk
        // starts past its end (sealed chunks are time-ordered).
        int64_t next_start = (cursor + 1 < sealed.size())
            ? sealed_chunk_start(sealed[cursor + 1]) : INT64_MAX;
        for (auto &kv : buckets)
            upsert_bucket(root, kv.first, kv.second, step.aggregate,
                          next_start >= kv.first + target_us);

        cat.set_processed_cursor(step_name, target_us, cursor + 1);  // durable mark
        if (cursor % kCheckpointEvery == 0) persist_superblock();
    }
    persist_superblock();
}

// Two-phase bucket node inside the step's B-tree:
//   open   -> append raw samples (chunk contributes exactly once thanks to cursor)
//   sealed -> wire_func_t evaluated once over all samples, raw samples dropped
static void upsert_bucket(block_id_t root, int64_t start_us, const bucket_accum_t &acc,
                          const wire_func_t &aggregate, bool cover_complete) {
    scoped_ptr_t<buf_lock_t> buf = btree::acquire(root);   // RMW upsert
    bucket_node_t node = bucket_exists(buf, start_us) ? read_bucket(buf, start_us)
                                                      : bucket_node_t{start_us};
    node.samples.insert(node.samples.end(), acc.samples.begin(), acc.samples.end());
    node.sample_count += acc.samples.size();

    if (cover_complete) {
        counted_t<const ql::func_t> fn = aggregate.compile();   // wire -> compiled
        node.value = fn->call(nullptr, ql::datum_t::make_array(node.samples));
        node.sealed = true;
        node.samples.clear();                       // final value only
    }
    write_bucket(buf, node);
}

// ---- store.cc, store_t::do_read rget branch : planner auto-selection ----
if (read.type == read_type_t::RGET) {
    const auto &rget = boost::get<read_t::rget_read_t>(read);
    // NEW: planner-selected downsample path.
    if (rget.downsample.has_value()) {
        const auto &ds = *rget.downsample;
        const downsample_root_t *r =
            catalog_.get_downsample_root(ds.step_name, ds.target_interval_us);
        if (r == nullptr) return empty_rget_response();   // not built yet / removed
        // rget.ts_range prunes bucket keys exactly like it prunes chunk roots today:
        //   left  = bucket_of(ts_range.left.left,  target)
        //   right = bucket_of(ts_range.right.left, target)
        return read_downsample_btree(r->root, ds, rget, batchspec, ...);
    }
    // existing chunk-root path unchanged (rget.ts_range pruning)
}

// ---- ops.hpp / terms/downsample.cc : term lowering (planner hook) ----
// table.downsample(step)  ->  rget_read_t with .downsample spec
counted_t<val_t> downsample_term_t::eval_impl(scope_env_t *env, args_t *args, ...) {
    counted_t<table_t> tbl = args->arg(env, 0)->as_table();
    datum_t spec = args->arg(env, 1)->as_datum();   // {step, interval, agg}
    return tbl->downsample_read(spec, env);          // lowers to rget; store.cc prunes
}

// ---- config change path : enqueue the background pass (validated step already persisted) ----
if (step_was_added) {
    for (const auto &shard : table->shards())
        job_manager->add_job(job_t(job_type_t::DOWNSAMPLE,
            downsample_job_data_t{table_id, step_name},
            job_priority_t::LOW, /*cancelable*/ true));
}