Deep Dive: Clustering¶

This document describes HyperbyteDB's clustering subsystem: node state machine, cluster bootstrap, startup and reconnect sync, Raft consensus for schema mutations, write replication, the replication log, and graceful drain.

Table of Contents¶

1. Architecture Overview
2. Node State Machine
3. Cluster Bootstrap
4. Startup Sync
5. Reconnect Sync
6. Raft Consensus
7. Write Replication
8. Mutation Replication
9. Replication Log
10. Cluster Convergence
11. Graceful Drain
12. Peer Internal Endpoints
13. Cluster Configuration
14. Metrics
15. Series Sharding (Experimental)

1. Architecture Overview¶

HyperbyteDB uses a hybrid replication model:

• Data writes are replicated to all peers (master-master). Each node independently accepts writes and fans them out via HTTP. The fan-out is per-node configurable as either async (fire-and-forget; default) or sync_quorum (await W-of-N peer acks before responding to the client). See Section 7 -- Write Replication.
• Schema mutations (CREATE/DROP DATABASE, DELETE, user management, CQ management) are routed through Raft consensus (via openraft) to ensure consistent ordering across the cluster. Schema replication uses a separate code path and is not affected by the data replication mode.

            +---------+     +---------+     +---------+
            | Node 1  |<--->| Node 2  |<--->| Node 3  |
            +---------+     +---------+     +---------+
               |                |                |
               v                v                v
          Write repl.      Write repl.      Write repl.
          (async HTTP)     (async HTTP)     (async HTTP)
               |                |                |
               +-------+-------+-------+--------+
                       |               |
                  Raft (schema mutations)
                       |
              openraft consensus

Key properties¶

• Every node is readable and writable -- clients can connect to any node.
• Write replication is fire-and-forget -- local write succeeds immediately; replication is best-effort with retries.
• Schema mutations are consistent -- Raft ensures all nodes apply mutations in the same order.
• Self-healing via WAL catch-up and metadata sync on startup/reconnect (see sections 4–5).
• Graceful drain ensures no data loss when removing a node.

2. Node State Machine¶

File: src/domain/cluster/membership.rs

Each node in the cluster transitions through the following states:

  Joining --> Syncing --> Active --> Draining --> Leaving
                 ^           |
                 |           v
                 +---- Disconnected

States¶

State transitions¶

• Joining -> Syncing: After sending a join request to an active peer.
• Syncing -> Active: After successful data synchronization.
• Active -> Disconnected: When heartbeat misses exceed heartbeat_miss_threshold.
• Disconnected -> Syncing: When the node reconnects and needs to catch up.
• Active -> Draining: When a graceful shutdown is initiated.
• Draining -> Leaving: After the drain procedure completes.

ClusterMembership¶

pub struct ClusterMembership {
    pub version: u64,                      // Monotonic version counter
    pub nodes: HashMap<u64, NodeInfo>,     // node_id -> NodeInfo
}

pub struct NodeInfo {
    pub node_id: u64,
    pub addr: String,
    pub state: NodeState,
    pub joined_at: i64,
    pub last_heartbeat: i64,
    pub needs_sync: bool,
}

• version is bumped on every mutation (add_node, remove_node, set_state).
• SharedMembership = Arc<RwLock<ClusterMembership>> provides thread-safe shared access.

Key methods¶

3. Cluster Bootstrap¶

File: src/application/cluster/bootstrap.rs

ClusterBootstrap::init(config)¶

Called during application startup when cluster.enabled = true.

1. Create replication log directory if it doesn't exist.
2. Open ReplicationLog -- RocksDB-backed store for WAL and mutation ack tracking.
3. Initialize membership -- create ClusterMembership with the local node as Active.
4. Build PeerClient -- HTTP client for replication, configured with the peer list.
5. Derive peer_addrs from config.cluster.peers (comma-separated).

start_raft(config, metadata)¶

1. Open RaftStore (RocksDB with meta, logs, state column families) in config.raft_dir.
2. Build Raft Network (HTTP transport).
3. Build openraft::Raft<TypeConfig> with configured heartbeat interval, election timeout, and snapshot threshold.
4. If node_id == 1: Initialize the Raft cluster with the local node as the sole voter.
5. Return HyperbytedbRaft.

run_startup_sync(config, metadata, wal, data_dir)¶

1. Set node state to Syncing.
2. Build a SyncClient.
3. Determine if this is a new node (no existing data) or reconnect:
4. New node: Call sync_client.join_and_sync().
5. Existing node: Call sync_client.reconnect_sync().
6. Retry up to 5 times on failure.
7. On success: set state to Active.
8. On failure after retries: set state to Active with needs_sync = true (allows the node to operate but flags it for later re-sync).

4. Startup Sync¶

File: src/adapters/cluster/sync_client.rs

join_and_sync() -- Full join flow for new nodes¶

1. Pick a sync peer via pick_sync_peer():
2. First active peer from membership.
3. If none, probe fallback addresses from config.
4. Send join request -- POST /internal/membership/join with JoinRequest { node_id, addr }.
5. Get manifest — GET /internal/sync/manifest from the peer.
6. Returns SyncManifest { node_id, wal_last_seq, databases } with database/RP/measurement definitions.
7. Sync metadata — GET /internal/sync/metadata:
8. Imports databases, measurements, users, tombstones, and CQ definitions.
9. WAL catchup — GET /internal/sync/wal:
10. Stream WAL entries from the peer starting from the local sequence.
11. Append each entry to the local WAL; flush builds local MergeTree tables.
12. Set state to Active.

5. Reconnect Sync¶

File: src/adapters/cluster/sync_client.rs

reconnect_sync() -- For existing nodes reconnecting after downtime¶

The reconnect strategy is chosen based on the WAL sequence gap between the local node and the peer:

Gap Analysis:
  local_wal_seq vs peer_wal_seq
       |
       v
  +-------------------+
  | gap == 0          | --> nothing to do
  | gap <= 10,000     | --> WAL catchup via /internal/sync/wal
  | gap > 10,000      | --> Full metadata sync + WAL catchup
  +-------------------+

Gap == 0 (no missed writes)¶

No action required.

Small gap (<= 10,000 entries)¶

WAL catchup from the peer (GET /internal/sync/wal), then mark Active.

Large gap (> 10,000 entries)¶

Full metadata sync plus WAL catchup (same metadata/WAL endpoints as startup join).

6. Raft Consensus¶

Files: src/adapters/cluster/raft/

HyperbyteDB integrates openraft for schema mutation consensus.

Type configuration¶

// src/adapters/cluster/raft/mod.rs
struct TypeConfig;
impl openraft::RaftTypeConfig for TypeConfig {
    type D = ClusterRequest;     // Log entry data
    type R = ClusterResponse;    // Apply result
    type NodeId = u64;
    type Node = BasicNode;
    type SnapshotData = Cursor<Vec<u8>>;
    type AsyncRuntime = TokioRuntime;
}

pub type HyperbytedbRaft = openraft::Raft<TypeConfig>;

Cluster request types¶

// src/adapters/cluster/raft/types.rs
pub enum ClusterRequest {
    SetNodeState { node_id: u64, state: NodeState },
    SchemaMutation(MutationRequest),
    ShardMapMutation(Box<ShardMapOp>),  // when [sharding] enabled
}

pub struct ClusterResponse {
    pub ok: bool,
    pub message: String,
}

RaftStore (log storage)¶

File: src/adapters/cluster/raft/log_store.rs

RocksDB-backed storage with three column families:

Implements RaftStorage, RaftLogReader, and RaftSnapshotBuilder.

State machine application¶

File: src/adapters/cluster/raft/state_machine.rs

When a ClusterRequest is committed by Raft:

• SetNodeState: Updates the local SharedMembership.
• SchemaMutation: Calls the appropriate MetadataPort method:
• CreateDatabase -> metadata.create_database()
• DropDatabase -> metadata.drop_database()
• Delete -> metadata.store_tombstone()
• CreateUser -> metadata.create_user()
• etc.
• ShardMapMutation: Applies a ShardMapOp to the local shard map (bootstrap, split, merge, TransferPrimary, MovePeer, etc.). See Section 15.

Raft-to-membership synchronization¶

sync_raft_membership_to_shared() propagates Raft membership changes to the local SharedMembership. Raft voters are mapped to Active or Syncing states.

Network transport¶

File: src/adapters/cluster/raft/network.rs

Raft messages are exchanged via HTTP:

Configuration¶

7. Write Replication¶

Files: src/application/peer_ingestion_service.rs, src/adapters/cluster/peer_client.rs

Per-node modes¶

[cluster.replication] selects this node's coordinator behavior. Receivers always serve both styles from the same /internal/replicate endpoint, so any combination of modes across the cluster is safe.

Async flow (mode = "async")¶

1. Client writes to any node.
2. PeerIngestionService::ingest():
3. Parse line protocol, register metadata, append to local WAL.
4. Dispatch to PeerClient::replicate_write() (spawns background task).
5. Return 204 No Content to the client.
6. PeerClient::replicate_write():
7. Push the batch onto the bounded outbound coalescer.
8. The outbound loop coalesces consecutive batches with the same (db, rp, precision) and contiguous wal_seq (up to replication_max_coalesce_body_bytes).
9. For each peer, spawn a task that POSTs to /internal/replicate with the line-protocol body, X-Hyperbytedb-* routing headers, and X-Hyperbytedb-Replicated: true.
10. On success (2xx): replication_log.set_wal_ack(peer_id, wal_seq).
11. On failure: retry with exponential backoff; on exhaustion, push to hinted handoff.

Sync quorum flow (mode = "sync_quorum")¶

1. Client writes to any node. PeerIngestionService::ingest() parses, registers metadata, and appends to the local WAL.
2. Resolve required = min_acks.resolve(active_peers().len()). If required == 0 (single-node cluster), return Ok(()) immediately.
3. Acquire one shared inflight permit from batch_semaphore (sync and async share the same backpressure budget).
4. Spawn one per-peer task; each sets X-Hyperbytedb-Sync: true and parses the peer's 200 OK body for ack_seq. Per-peer retry/backoff is identical to the async path.
5. Coordinator awaits acks via a select_all loop with deadline ack_timeout_ms:
6. Once required peers ack, return 204 No Content to the client.
7. On timeout, return 504 Gateway Timeout (ReplicationQuorumTimeout) -- but the in-flight peer tasks continue so durability still ratchets forward in the background.
8. Receiver: same ReplicationApplyQueue as the async path; the only difference is the handler awaits the apply oneshot before responding (see handle_replicate_write in src/adapters/http/peer_handlers.rs).

Retry behavior (both modes)¶

• Base delay: 1 second.
• Max delay: 30 seconds.
• Max retries: replication_max_retries (default 5).
• Backoff: min(delay * 2, 30s).
• On exhaustion: push to hinted handoff (when configured) and return failure.

Mixed-mode and rolling restart¶

The mode is per-node coordinator config. Receivers ignore the mode entirely and react only to X-Hyperbytedb-Sync. Concretely:

• [A=sync_quorum, B=async, C=async]: A's writes await acks from B and C; B and C's writes are async as today. Safe.
• Rolling-restart node A out of sync_quorum: B and C in sync_quorum recompute required against the smaller active_peers() count and continue serving (e.g. 1 required peer ack instead of 2 in a 3-node cluster).
• A new sync_quorum coordinator talking to an old binary that ignores X-Hyperbytedb-Sync still works -- the old peer 200s on success, which counts as one ack.

Loop prevention¶

The receiving node checks for X-Hyperbytedb-Replicated: true. If present, it: - Persists the write locally (WAL + metadata). - Does NOT re-replicate to other peers.

8. Mutation Replication¶

File: src/adapters/cluster/peer_client.rs

With Raft¶

When Raft is configured, schema mutations are routed through Raft consensus:

raft.client_write(ClusterRequest::SchemaMutation(mutation_request)).await

Raft ensures all nodes apply the mutation in the same order.

Without Raft (fallback)¶

Mutations are broadcast directly to all peers:

1. Append to the local replication log: replication_log.append_mutation(request) -> returns seq.
2. For each active peer:
3. POST /internal/replicate-mutation with MutationReplicateRequest { seq, origin_node_id, mutation }.
4. On success: replication_log.set_mutation_ack(peer_id, seq).
5. On failure: retry with exponential backoff.

MutationRequest variants¶

pub enum MutationRequest {
    CreateDatabase(String),
    DropDatabase(String),
    CreateRetentionPolicy { db: String, rp: RetentionPolicy },
    CreateUser { username: String, password_hash: String, admin: bool },
    DropUser(String),
    Delete { database: String, measurement: String, predicate_sql: String },
    CreateContinuousQuery { database: String, name: String, definition: ContinuousQueryDef },
    DropContinuousQuery { database: String, name: String },
}

Deduplication¶

The receiving node calls replication_log.check_and_record_mutation(origin_node_id, seq). This returns true only if seq is greater than the last applied sequence for that origin, preventing duplicate application.

9. Replication Log¶

File: src/adapters/cluster/replication_log.rs

The replication log is a RocksDB-backed store that tracks WAL and mutation acknowledgements from peers.

Key layout¶

Key operations¶

Safe WAL truncation¶

The flush service uses min_wal_ack() to determine how far the WAL can be safely truncated:

safe_truncate = min(chunk_max_seq, min_wal_ack_across_peers)

This ensures peers that are catching up can still read needed WAL entries.

10. Cluster Convergence¶

Peers stay aligned through three mechanisms:

1. Write replication — after local WAL append, the coordinator fans out line protocol to peers (ReplicationPort / PeerClient).
2. Startup and reconnect sync — SyncClient exchanges metadata snapshots and WAL tail entries via /internal/sync/{manifest,metadata,wal}.
3. Local flush — each peer replays its WAL (including replicated entries) into local MergeTree tables via PointsSinkPort.

There is no separate file-repair loop. chDB storage on each node is derived from the shared WAL + metadata contract.

11. Graceful Drain¶

File: src/application/cluster/drain.rs

The drain procedure ensures no data loss when removing a node from the cluster.

DrainService::drain()¶

Step 1: Set node state to Draining
        (HTTP write handler starts rejecting writes with 503)
        |
Step 2: Flush all WAL entries to chDB MergeTree tables
        (flush_service.drain() via FlushPort -- loops until WAL is empty)
        |
Step 3: Wait for replication acks (up to 60 seconds)
        (loop: check if all peers have acked local WAL seq
         and mutation seq, sleep 2s between checks)
        |
Step 4: Notify peers of leave
        (POST /internal/membership/leave to all active peers)
        |
Step 5: Set node state to Leaving

(There is no Merkle verify step. Any tail divergence is reconciled by per-file CRC diff repair on the remaining peers.)

Step details¶

Step 1 -- Reject writes: The HTTP write handler checks node state and returns 503 Service Unavailable with an optional X-Hyperbytedb-Redirect header for Draining, Leaving, Syncing, or Joining states.

Step 2 -- Flush WAL: flush_service.drain() repeatedly calls flush() until no WAL entries remain. This ensures all ingested data is persisted in native MergeTree tables.

Step 3 -- Wait for acks: Polls replication_log.get_wal_ack(peer_id) and get_mutation_ack(peer_id) for each active peer. Waits until all peers have acked up to the local WAL and mutation sequence numbers. Times out after 90 seconds with a warning.

Step 4 -- Notify peers: Sends POST /internal/membership/leave to all active peers so they remove this node from their membership.

Step 5 -- Set Leaving: Final state transition. The node can be safely shut down.

12. Peer Internal Endpoints¶

Files: src/adapters/http/peer_handlers.rs, src/adapters/http/raft_handlers.rs

Replication endpoints¶

Membership endpoints¶

Sync endpoints¶

Cluster consistency after replication: WAL replication plus metadata/WAL sync on join/reconnect. Each peer builds local MergeTree tables by flushing replicated WAL entries.

Raft endpoints¶

Drain endpoint¶

Shard endpoints (when [sharding] enabled = true)¶

Shard handlers validate region_id, epoch, and local peer membership before applying data-plane work. Stale epochs return 409 Conflict.

13. Cluster Configuration¶

File: src/config.rs -- ClusterConfig

14. Metrics¶

Replication metrics¶

Cluster state metrics¶

15. Series Sharding (Experimental)¶

When [sharding] enabled = true (requires [cluster] enabled = true), HyperbyteDB partitions each measurement's series space by series_id into regions covering [0, u64::MAX). Each region has a primary and replica peers; data is not fully replicated to every cluster node.

Client /write or /query (any node)
        |
        v
+----------------------------------+
| Coordinator (local node)         |
|  - Resolve region via shard map  |
|  - Local peer: apply directly    |
|  - Remote region: scatter HTTP   |
+----------------------------------+
        |
        +--> /internal/shard/write   (writes)
        +--> /internal/shard/query   (SELECT)
        +--> /internal/shard/metadata (SHOW *)
        +--> /internal/shard/delete  (DELETE cleanup)

Configuration keys are documented in Configuration Reference.

Control plane¶

• Shard map is stored in RocksDB (RocksDBShardMap) and replicated via Raft (ClusterRequest::ShardMapMutation).
• The Raft leader runs a shard scheduler (ShardScheduler) that evaluates region heartbeats and proposes split, merge, rebalance, and primary-transfer operations.
• Split/merge respect split_merge_interval_secs cooldown and schedule_limit concurrent ops.
• max_regions_per_measurement caps runaway splits; load_split_qps_threshold optionally triggers load-based splits when non-zero.
• region_id is globally unique across all measurements in the shard map. Bootstrap allocates via monotonic next_region_id (build_bootstrap_op in shard_routing.rs); splits allocate the right-hand child the same way. apply_shard_map_op rejects duplicate IDs on bootstrap and split; MeasurementShardSpace::validate rejects duplicates within a space; ShardMap::validate_global_region_ids checks the full map. Heartbeats, epoch CAS, and lookup_region_by_id all assume this invariant — reusing an ID causes endless stale_epoch rejections.

Data plane — writes¶

• Incoming points are bucketed by series_id using ShardLocationCache.
• Points for regions where this node is an Active peer are written locally (WAL + region-scoped replication).
• Other regions are forwarded via scatter_to_region_peers to /internal/shard/write (primary first, then other Active replicas).
• Region-scoped replication sets target_node_ids on outbound batches so only Active region peers receive WAL fan-out.

Data plane — queries¶

• SELECT queries translate TimeseriesQL to ClickHouse SQL, select overlapping regions, inject per-region series_id range predicates, then scatter to Active peers via /internal/shard/query. Results are merged on the coordinator.
• SHOW TAG KEYS/VALUES/SERIES scatter via /internal/shard/metadata and merge on the coordinator.
• DELETE / DropSeries / DropMeasurement store a logical tombstone once, then fan out physical cleanup via /internal/shard/delete (primary-only; skipped until primary is Active or failovered).

Scatter and down-node avoidance¶

File: src/application/shard_routing.rs, src/application/shard_peer_resolution.rs

scatter_to_region_peers is the shared retry loop for queries, metadata, and write forwards:

1. Build an ordered candidate list: self (if Active and in region.peers) → primary (if Active) → other Active peers.
2. Exclude non-Active membership states (Disconnected, Draining, etc.).
3. Truncate to scatter_max_peer_attempts remote tries (self is handled separately on the coordinator).
4. For each candidate, invoke the caller's HTTP callback with scatter_peer_timeout_ms.
5. On success after the first remote candidate, increment hyperbytedb_shard_scatter_fallback_total{kind}.
6. StaleShardEpoch aborts immediately without trying further peers (map changed underfoot).
7. If all candidates fail, return PeerUnreachable.

Cluster heartbeats probe /health (readiness) in addition to /ping; peers that fail readiness are marked non-Active and drop out of scatter routing.

Replica reads are eventually consistent — a scatter query may succeed against a replica while the primary is down, but lagging replicas can return slightly stale data until replication catches up.

Primary failover¶

When a region primary stays non-Active longer than primary_failover_after_secs, the Raft leader's scheduler proposes TransferPrimary to the next Active peer in the region. Metrics: hyperbytedb_shard_primary_failover_total and hyperbytedb_shard_primary_failover_skipped_total{reason}.

Remote DELETE cleanup waits for an Active primary; use drain or wait for automatic failover before expecting cross-node delete cleanup during a primary outage.

Transfer and lifecycle¶

Region transfers move vacated series_id ranges before split/map mutations commit so clients never read an empty new owner:

1. Push — source exports WAL tail + flushed chDB rows in chunked /internal/shard/transfer payloads (transfer_id, seq, done).
2. Split / rebalance / merge — Raft commits the shard-map change (epoch CAS on Split/Merge).
3. Ack — source sends Ack to the destination primary; destination purges stale materialized-view partials for the vacated range.
4. Drop — source deletes range-scoped series metadata and issues chDB DELETE for the vacated fact/series tables.

The scheduler holds a per-region operator lock while split/merge/rebalance/failover/transfer runs; schedule_limit counts live operators, not per-tick slots.

• Drain performs transfer, TransferPrimary, then MovePeer (re-reads epoch after primary transfer).
• Sync manifests include per-region WAL watermarks; failover prefers the most caught-up Active peer (leader eligible).

Split-transfer reconciliation¶

A split commits via Raft before its historical-row re-push completes. If the destination's Raft apply lags the leader's, re-home/push RPCs answer 404 ("region range not found") or 409 (stale epoch). These failures are not lost:

• Apply-time normalization records durable intent on the child region: transfer_verified=false (+ transfer_first_seen) whenever a split child's primary differs from its parent's. The flag rides the Raft-committed map, so leadership change or process restart rebuilds the work queue from a map scan.
• The leader keeps an in-memory reconciliation queue (ordering detail only) and drains it each tick. Movement retries use authoritative sources only — the old primary recorded at split time, then live peers with a non-zero region WAL watermark (a data-less peer would verify vacuously).
• Satisfaction requires positive proof: every exported point confirmed applied at the destination. Sentinel outcomes (transfer_id == 0, i.e. source==destination collision; empty exports) never clear debt.
• On verified movement the leader proposes the ClearVerified shard-map op. Disable with [sharding] transfer_clear_proposals_enabled = false on mixed-version clusters: nodes older than the op cannot decode it from the Raft log.
• Splits and merges are refused while either region carries outstanding debt. Stuck entries escalate once via TransferPrimary to another live peer (stage-before-commit ordering prevents committing a data-less owner), then park with hyperbytedb_shard_transfer_parked_total after ~10 stalled ticks past a minimum age — operator action required.
• Materialized-view rollup destinations (SummingMergeTree) are excluded from reconciliation entirely: re-delivery would permanently double additive aggregates. Their provisioning flows through the MV backfill path.
• Raw measurement tables are ReplacingMergeTree(ingest_seq) keyed by (series_id, time): retried pushes dedupe eventually, but a transient duplicate window exists until background merges collapse rows.

Membership self-truth (/health body)¶

Every /health response (200 and 503 alike) carries machine-readable self-readiness fields:

{"status": "...", "message": "...", "state": "active", "needs_sync": false}

Cluster peers parse these fields in their periodic probes and converge their local membership view toward each peer's self-reported state: a peer recorded Syncing is promoted to Active exactly when it reports state=active && needs_sync=false (caught up and accepting data), and demoted back when it reports syncing. Overall HTTP status never drives membership transitions — a wedged WAL batcher fails /health without changing what peers believe about regional catch-up. Peers that do not answer at all transition to Disconnected (the crash detector); responses without parseable bodies (legacy binaries) carry no information in either direction.

Region heartbeats¶

Store nodes report to the Raft leader via /internal/shard/heartbeat:

Distributed aggregate support¶

Multi-region fan-out rewrites SELECT into per-region partial aggregations, then merges on the coordinator:

Global LIMIT, OFFSET, and ORDER BY apply on the coordinator after merging (per-region SQL fetches limit+offset rows).

• After split, the scheduler runs region transfer before proposing the map change.
• Drain performs transfer + TransferPrimary + MovePeer for regions where the draining node is primary.
• Sync manifests include per-region WAL watermarks; join sync pulls from region primaries where the joining node is a peer.

Materialized views¶

When sharding is enabled, materialized views use the same per-node ClickHouse MV model as non-sharded clusters, with three extra behaviors:

1. Incremental triggers stay local. Source writes are already routed by series_id region, so each node installs fact + series ClickHouse MVs that fire on locally ingested source rows. No cross-region predicate is injected into MV SQL.

2. Destination queries fan out to all regions and all Active peers in each region. Materialized destination measurements store region-local partial rollups. Coordinators select every region in the dest shard map (no series_id predicate), query every Active peer in each region via /internal/shard/query, then merge partial rows with SummingMergeTree semantics (merge_materialized_rollup_results). Tag-subset GROUP BY (for example GROUP BY time(1m), "host") relies on this merge at read time.

3. WITH BACKFILL scatters before MV DDL. On CREATE MATERIALIZED VIEW ... WITH BACKFILL, the leader ensures dest schema, inserts historical fact + series rows per source region through /internal/shard/mv-backfill (with region series_id predicates), then installs ClickHouse MV objects without dropping the backfilled destination tables.

On create, both source and destination measurements are bootstrapped into the shard map. MV definitions replicate through Raft like other schema mutations; each node reconciles local ClickHouse MV objects on startup.

After a region transfer, destination partial rows sourced from transferred series_id ranges are purged on the ACK phase so stale rollups do not linger on vacated peers.

Metrics¶

Clean cluster runbook (kind, 6-node)¶

Use this sequence before sharding acceptance runs (e2e G0–G9) or any test that depends on a pristine shard map. Do not enable sharding on PVCs that already contain bootstrapped measurements — leftover regions pollute G0/G1 and mask bootstrap bugs.

1. Fresh storage — ./deploy/kind/setup.sh hdb-down --sharded (or hdb-reset --sharded) deletes the HyperbytedbCluster CR, lets the operator garbage-collect StatefulSet/services/config, then removes PVCs. Do not scale the operator to 0 or hand-patch hyperbytedb-config.
2. Deploy 6-node cluster — ./deploy/kind/setup.sh up --sharded creates a 6-worker kind cluster and applies hyperbytedb-cr-6node-sharded.yaml with spec.sharding. For an existing kind cluster: ./deploy/kind/setup.sh hdb-reset --sharded.
3. Operator drives Raft — keep hyperbytedb-operator at 1 replica; it calls /cluster/membership/add-node as pods become Ready. Wait until /cluster/nodes reports 6 active members (setup script polls this).
4. Settle — wait ≥60s after pods are Ready and Raft is complete so shard heartbeats stabilize. hdb-up --sharded sleeps 60s by default (SHARDED_SETTLE_SECS).
5. Acceptance — run deploy/kind/run-sharding-e2e.sh (in-cluster Job).

Limitations¶

• Enable sharding only on new clusters; in-place conversion from full-copy replication is not supported.
• Proxy load balancing is not shard-aware; any Active node can coordinate scatter-gather queries and write forwards.
• sync_quorum on forwarded writes: /internal/shard/write honors the configured replication mode; SyncQuorum fails the request when region peer acks do not meet quorum.
• Split keys prefer the median local series_id in the region (metadata catalog scan), falling back to range midpoint when empty.
• percentile, median, mode, and spread are rejected on multi-region fan-out until a merge strategy exists.