pj_datastore/docs/ARCHITECTURE.md
Scope: this document covers the scalar
DataEnginepath — the columnar store, encoding, queries, andDerivedEngine. The library also compiles three first-class components documented elsewhere or summarized below: the opaque-blobObjectStore(its own doc,OBJECT_STORE_DESIGN.md), theColorMapRegistry+ its C-ABI host (§3 Color Map Layer), and the streaming two-engineflushTo/setTargetswap (§3 Plugin Host Layer).
Two libraries with a strict dependency direction:
pj_base (plotjuggler_sdk/pj_base/, from the SDK submodule): Vocabulary types and SDK headers with zero external dependencies. Defines:
Timestamp (int64_t, nanoseconds since Unix epoch)Range<T> inclusive min/max pairsDatasetId, TopicId, FieldId, SchemaId, TimeDomainId (all uint32_t), ChunkId (uint64_t), NodeId (uint32_t)PrimitiveType enum (12 variants: kFloat32..kUint64, kBool, kString)NumericType, NumericValue (variant of all numeric scalars)TypeTreeNode (recursive schema tree: primitive, struct, array, enum nodes)Span<T> (alias for std::span), BitSpan (bit-level view with offset)Expected<T> / Status for fallible operations, PJ_ASSERT for invariantspj_datastore (pj_datastore/): Engine implementation. External dependencies: fmt, tsl::robin_map, nanoarrow (Arrow IPC import).
Dependency: pj_datastore -> pj_base. No reverse dependency. pj_plugins depends on pj_base only, never on pj_datastore.
Hierarchy: Dataset -> Topic -> Chunk -> Column
Dataset (DatasetInfo): Identity (DatasetId), source_name string, bound TimeDomain. Represents one data source (file, live connection).
Topic (TopicDescriptor + TopicStorage): Named data stream within a dataset. Key fields: schema_id (0 = schemaless), max_chunk_rows (default 1024), array_expansion_limit (default 64). TopicStorage owns the committed chunk deque and column descriptors for schemaless topics.
Schema (TypeRegistry + TypeTreeNode): Tree of named typed nodes. TypeRegistry assigns SchemaId values, supports lookup by id/name, and additive-only evolution via evolveSchema(). Schemas are shared across topics.
Chunk (TopicChunk): Sealed, immutable storage unit produced by TopicChunkBuilder::seal(). Contains: ChunkId (monotonic atomic counter), timestamps vector, columns vector (each holding EncodedData + optional BitVector validity bitmap + shared_ptr<ColumnDescriptor>), and ChunkStats (t_min, t_max, row_count, per-column ColumnStats).
Column (TopicChunk::Column): The physical unit. EncodedData is a variant of 5 encoding types (see section 5). ColumnDescriptor carries field_id, logical_type, and fully-qualified field_path (e.g. "pose.position.x").
DataEngine — Central owner of all state. Stores datasets, topics (as TopicStorage), time domains, and the global TypeRegistry, all in hash map containers (tsl::robin_map internally, std::unordered_map in headers). Provides:
createDataset(), createTopic(), createTimeDomain() with monotonic ID allocation or explicit caller-requested IDscommitChunks() — appends sealed chunks to TopicStorage, returns deduplicated list of changed TopicIdsenforceRetention() — evicts old chunks across all topicscreateWriter() and createReader()TypeRegistry — registerSchema() assigns a new SchemaId. registerOrGet() returns an existing ID if the name matches (for late-discovery schemas). evolveSchema() validates additive-only changes (no field removal or type change).
TopicChunkBuilder — Mutable builder that accumulates rows for one topic. Two append paths:
beginRow(timestamp) -> set<T>(col, value) / setNull(col) -> finishRow(). finishRow() pads unset columns with null.appendTimestamps(span) -> appendColumn<T>(col, span) per column -> appendColumnValidity(col, bitspan) -> finishBulkAppend() (computes stats).Tracks per-column ColumnStats incrementally (min, max, null_count, is_constant, run_count). seal() encodes columns, assigns a monotonic ChunkId via std::atomic<ChunkId>, and produces an immutable TopicChunk.
TopicStorage — Per-topic container of committed chunks in a std::deque<TopicChunk>. appendSealedChunk() appends in commit order without rejecting any chunk — out-of-order ingest means chunk time ranges may overlap, so queries merge across them (see §5). evictBefore() removes the contiguous prefix of chunks whose t_max < threshold and raises a per-topic retention floor. Also stores column descriptors for schemaless (schema_id == 0) topics and per-field array expansion counts.
TypedColumnBuffer — In-memory typed column buffer used internally by the builder. One buffer per column. Supports per-type single-row append (appendFloat64, appendInt64, etc.) and bulk append (appendFloat64Bulk, etc.). String storage uses a separate offsets buffer (RawBuffer) with Arrow-compatible uint32 offset layout. Validity bitmap (BitVector) is lazily initialized -- only allocated when the first null is appended.
RawBuffer — Growable byte vector wrapping std::vector<uint8_t>. Used for column value storage and encoded payloads.
BitVector — Owning packed validity bitmap with Arrow-compatible LSB-first layout. Supports initValid(), setNull(), isValid(), countNulls(), and bulk assignBytes().
DataWriter — High-level write facade bound to a DataEngine. Manages one TopicChunkBuilder and a pending chunk list per topic. Key operations:
registerTopic() / registerScalarSeries() — topic creationensureColumn() — dynamic column addition for schemaless topics. Rejects if a row is in progress (for new columns only). Seals any pending builder before modifying column layout.expandArray() — variable-length array expansion. Seals current builder, adds new ColumnDescriptors, updates TopicStorage. Clamps to array_expansion_limit.appendColumns() — bulk ingest with auto-chunking: splits batches that exceed max_chunk_rows, calling appendTimestamps / appendColumn / finishBulkAppend / autoSeal per sub-batch.flush() / flushAll() — seals remaining builders and returns vector<pair<TopicId, TopicChunk>> for commitChunks().autoSeal() — called when builder is full (rowCount >= max_chunk_rows). Seals builder, moves chunk to pending_chunks_.DataReader — Read-only facade over committed DataEngine storage. Provides listDatasets(), listTopics(), getTypeTree(), getMetadata(), rangeQuery(), latestAt(), and series(topic_id, column_index).
RangeCursor — Iterates rows in [t_min, t_max] across the chunk deque. Chunk time ranges may overlap (out-of-order ingest), so the constructor seeds one frontier per intersecting chunk into a min-heap on (timestamp, chunk index). Supports:
forEach(callback) — per-row iteration via SampleRow (timestamp + chunk pointer + row index), in globally ascending timestamp order even across overlapping chunks; the heap advance is O(1) while chunks don't actually overlapforEachChunk(callback) — bulk iteration via ChunkRowRange (chunk pointer + row start/end); runs are per-chunk sorted but delivered in commit order, so under overlap they may interleave in time — bulk consumers tolerate that or use forEachlatestAt(chunks, t) — Scans candidate chunks (per-chunk binary search) for the most recent row at or before t; later-committed chunks win timestamp ties. Returns optional<SampleRow>.
Both are free functions operating on const std::deque<TopicChunk>&.
SeriesReader — Views one numeric/bool topic column as a virtual vector of (timestamp, value) samples. It is bound to a topic and column when created through DataReader::series(). Null physical rows and chunks where the column does not exist are skipped by definition. Provides:
size() / empty() — sample count, not physical row countsampleAt(index) — lookup by virtual series indexsampleAtOrBeforeTime(t) / sampleAtOrAfterTime(t) — lookup by timestamp over value-bearing samplessamples(Range<Timestamp>) — cursor over samples in an inclusive time rangebounds() / bounds(Range<Timestamp>) — valid time and value ranges for the seriesSeriesCursor — Iterates value-bearing samples in [time.min, time.max] across chunks. It returns SeriesSample values with timestamp, double value, chunk pointer, and physical row index for low-level consumers that need provenance.
Seven StorageKind values map logical PrimitiveType to physical storage:
| StorageKind | Source PrimitiveTypes |
|---|---|
| kFloat32 | kFloat32 |
| kFloat64 | kFloat64 |
| kInt32 | kInt32 |
| kInt64 | kInt8, kInt16, kInt64 |
| kUint64 | kUint8, kUint16, kUint32, kUint64 |
| kBool | kBool |
| kString | kString |
Five EncodingType values, selected at seal time per column:
| EncodingType | When used | Representation |
|---|---|---|
kRaw (RawBuffer) | Default for float/uint64 when not constant | Typed byte buffer |
kConstant (ConstantEncoded) | All non-null values equal (is_constant && rowCount > 0) | 8-byte value + count |
kFrameOfReference (FrameOfReferenceEncoded) | kInt32/kInt64 when range fits in fewer bytes | int64 reference + packed uint8/16/32 offsets |
kDictionary (DictionaryEncoded) | Always for kString | Unique string list + narrowed uint8/16/32 indices |
kPackedBool (PackedBools) | kBool when not constant | 1 bit per value, LSB first |
EncodedData = std::variant<RawBuffer, ConstantEncoded, FrameOfReferenceEncoded, DictionaryEncoded, PackedBools>
Encoding selection in TopicChunkBuilder::seal():
dictionaryEncodeStrings().is_constant, otherwise packed bits via packBools().offsetBytesFor(range) < storageKindSize(kind), otherwise raw.is_constant, otherwise raw.DerivedEngine — Manages a transform DAG. Uses pimpl (DerivedEngineImpl). Key operations:
addSisoTransform() — registers a single-input/single-output node. Input must be a single-column topic. Creates an output scalar topic. Returns NodeId.addMimoTransform() — registers a multi-input/multi-output node. All inputs must be single-column. Creates N output topics.topologicalOrder() — Kahn's algorithm. Cycle detection via DFS at registration time.onSourceCommitted(changed_topics) — marks directly dependent nodes dirty.scheduleAll() — processes all dirty nodes in topological order (incremental path).scheduleActive(active_nodes) — processes only specified nodes and their transitive upstream dependencies.recomputeBatch(node_id) — clears output topic, calls transform.reset(), replays full input history.ISISOTransform — Point-at-a-time interface. calculate(time, input, &out_time, &out_value) -> bool. Called in strictly ascending timestamp order. State persists across chunk boundaries. reset() clears state for batch recompute. outputKind() declares output StorageKind (default kFloat64).
IMIMOTransform — N inputs -> M outputs. calculate(time, inputs_span, &out_time, &outputs_vec) -> bool. Exact-timestamp inner join: only called when ALL input topics have a sample at the same timestamp. outputKinds(input_kinds) declares one StorageKind per output topic.
VarValue = std::variant<int64_t, uint64_t, double, std::string> — Universal value type for transform I/O. Mapping: float32/float64 -> double, int8..int64/bool -> int64_t, uint64 -> uint64_t, string -> std::string.
Incremental scheduling: each node tracks a last_processed_chunk_id watermark. scheduleAll() iterates only chunks with id > watermark, reads each row, calls calculate(), writes output via beginRow/set/finishRow, then flushes and commits.
Out-of-order input: transforms have a strict ascending-timestamp contract, so before running a node the scheduler checks whether any not-yet-processed input chunk lands at or before the node's timestamp watermarks (siso_last_ts for SISO; mimo_last_chunk_id + mimo_last_ts for MIMO). Such late input triggers recomputeBatch — reset transform state, clear outputs, fully replay over the time-merged input (the SISO replay feeds rows through the RangeCursor heap merge) — instead of incremental work.
ColorMapRegistry (colormap_registry.hpp) — Registry of named colormap callbacks. A plugin registers one or more named ColorMapEvalFns (scalar -> CSS color / #rrggbb) via registerMap() and selects an active one with setActive(); consumers (chart renderers, exporters) evaluate the active map per data point via evaluate(). The registry holds raw callback pointers + user contexts and does not own the plugin that supplied them.
makeColorMapRegistryHost() (colormap_registry_host.hpp) — Wraps a ColorMapRegistry as a C-ABI PJ_colormap_registry_t fat pointer for bind_colormap_registry. The fat pointer references the registry by address (the registry must outlive every bound plugin); the vtable is a static singleton, safe to share across plugins and threads.
DatastoreSourceWriteHost / DatastoreParserWriteHost / DatastoreToolboxHost — Bridge between the C ABI scalar-write protocol (PJ_source_write_host_t, etc.) and the C++ DataWriter/DataEngine. Each wraps a pimpl state struct. Provides raw() to get the C function-pointer table and flushPending() to seal/commit accumulated data.
The host translates C ABI calls (ensureTopic, ensureField, appendRecord) into DataWriter operations (registerTopic/ensureColumn, beginRow/set/finishRow).
Object hosts — DatastoreSourceObjectWriteHost, DatastoreParserObjectWriteHost, and DatastoreToolboxObjectReadHost are the ObjectStore peers of the scalar hosts (surfaces pj.source_object_write / pj.parser_object_write / pj.toolbox_object_read). They bridge the C ABI onto an ObjectStore rather than a DataEngine; see OBJECT_STORE_DESIGN.md for their write/read contracts.
Streaming two-engine swap — During paused streaming the host can route pushes to a secondary engine/store and swap back on resume. DataEngine::flushTo(dst) zero-copy-moves committed chunks from the secondary into the primary (topics matched by descriptor, per-topic monotonicity enforced); DatastoreSourceWriteHost::setTarget() / DatastoreParserWriteHost::setTarget() (and the object hosts' setTarget()) flush pending rows then atomically retarget the destination.
For the swap to be transparent to plugins that cache TopicHandle/FieldHandle (e.g. data_stream_dummy caches in onStart(); parser_protobuf caches on first message), both engines must assign the same TopicId and FieldId to a given (topic, field) — otherwise the cached id resolves to nothing on the post-swap target. The write hosts keep this lockstep themselves once the streaming manager wires a secondary via setSecondaryEngine(): every ensureTopic/ensureField — including the lazy column auto-create inside append* — is mirrored onto the secondary via DataEngine::createTopic(requested_id) and its field-level counterpart DataEngine::createTopicField(requested_id), forcing matching ids (the latter takes a std::optional<FieldId> so a forced id 0, the first field of any topic, is not confused with "auto-assign"). Mirroring is bidirectional, so a cached handle resolves on whichever engine is the active target. ObjectStore::flushTo() provides the matching blob-side move; bound ObjectTopicIds stay valid because each registerTopic is likewise mirrored at registration time.
Arrow IPC (PJ::arrow_import namespace):
schemaFromIpc() — parses Arrow schema from IPC stream bytes via nanoarrow. Maps Arrow fields to ArrowColumnMapping (arrow column index -> PJ column index + PrimitiveType). Unsupported Arrow types are skipped.importIpcStream() — reads record batches, appends via DataWriter::appendColumns() with validity bitmaps.writeHost.ensureTopic(name) -> host creates topic via DataWriter::registerTopic()writeHost.ensureField(topic, name, type) -> host calls DataWriter::ensureColumn()writeHost.appendRecord(topic, timestamp, fields) -> host calls DataWriter::beginRow(), set() per field, finishRow()finishRow() pads unset columns with null, updates per-column statsrowCount >= max_chunk_rows): autoSeal() seals builder, moves TopicChunk to pending_chunks_flushPending() -> DataWriter::flushAll() seals remaining builders -> DataEngine::commitChunks() appends to TopicStorageDataWriter::appendColumns(topic_id, timestamps, column_data_array)remainingCapacity()appendTimestamps(), appendColumn<T>() per column, finishBulkAppend() (computes stats), autoSeal() if fullDataReader::series(topic_id, column_index) validates the topic, column bounds, and numeric/bool value typeSeriesReader treats the column as a field-level time seriesSeriesSample has a valuebounds() returns min/max over value-bearing samples onlyDataReader::rangeQuery(QueryRange{topic_id, t_min, t_max}) -> RangeCursorforEach callback receives SampleRow (chunk pointer + row index)chunk->readNumericAsDouble(col, row), readString(), readBool(), isNull(), or batch via readColumnAsDoubles()engine.commitChunks() returns changed topic IDsderivedEngine.onSourceCommitted(changed_topics) -> marks dependent nodes dirtyderivedEngine.scheduleAll() processes dirty nodes in topological orderlast_processed_chunk_id), read each row, call transform.calculate(), write output via beginRow/set/finishRow, flush + commitlatestAt(), call transform.calculate() only when all inputs have matching timestampsbeginRow() / appendTimestamps() accept regressions — rejecting them silently lost multi-publisher data); seal() stable-sorts rows, so every sealed chunk is internally non-decreasing and duplicate timestamps keep arrival order.expandArray() seals the current builder before modifying column layout, preventing mid-chunk schema changes.TopicStorage::appendSealedChunk() accepts any chunk (deque stays commit-ordered). Queries merge across overlapping chunks; topic time extrema are scanned, never taken from the front/back chunks.TypedColumnBuffer only allocates a BitVector on first appendNull(). Sealed chunks include validity only when hasNulls() is true.readColumnAsDoubles() writes NaN at null positions, preventing confusion between null and zero.TopicChunkBuilder uses a static atomic<ChunkId> counter starting at 1. kInvalidChunkId (0) is the sentinel for "no chunk seen yet".DataEngine is guarded by one std::recursive_mutex (Impl::mutex_). The store
has exactly one reader thread (the Qt GUI: plot/catalog reads + DerivedEngine
recompute) and one-or-more ingest worker threads, so an exclusive lock is
equivalent to a reader-writer lock here and simpler. EVERY engine access — read or
write, on any thread — holds the lock via lockEngine(). DataWriter accumulates
in-memory and is not itself shared; TopicChunkBuilder::next_chunk_id_ remains a
std::atomic.
The mutex is recursive so the same thread may re-acquire it without
deadlocking: a held RangeCursor/SeriesReader whose consumer makes another read,
a compound mutator that calls another, or the write host's appendRecord →
ensureField. The one thing recursion does NOT make safe: calling a mutator from
inside a cursor forEach callback — that mutates the deque mid-iteration; don't.
Coordinated mechanisms:
topics is robin_map<TopicId, unique_ptr<TopicStorage>>,
so a createTopic rehash moves only pointers — a TopicStorage (and its
sealed_chunks_ deque) keeps its address for the engine's lifetime (topics are
never erased, only retired). A cached TopicChunk* survives a concurrent
createTopic.DataReader::rangeQuery/series take the lock
before the getTopicStorage lookup (rehash protection) and move it into
the returned RangeCursor/SeriesReader, which hold it for their whole (lazy
forEach) lifetime. SeriesCursor borrows its parent's lock. latestAt
materializes the row's values under the lock (no escaping TopicChunk*). The raw
getTopicStorage() accessor does not lock — a caller must hold lockEngine()
for the returned pointer's lifetime.WriteCore /
DataWriter, driven on the ingest worker) is a full engine client: each
write-host method (ensureTopic/ensureField/appendRecord/appendBoundRecord/
appendArrowStream/createDataSource, and flushPending) takes lockEngine()
at entry, so its getTopicStorage reads and its setColumnDescriptors /
builder mutations serialize against GUI reads. (TopicStorage has no internal
lock; the engine lock is its mutual exclusion.) TypeRegistry likewise has no
internal lock and is covered by the engine lock — every lookup/registerOrGet
site (DerivedEngine::addSisoTransform/addMimoTransform, the write host, the
reader) holds lockEngine(). The GUI bypass readers
CatalogModel::rebuildFromDatastore / PointSeriesXY take it too.commitChunks only push_backs, so a
worker append never invalidates a live cursor or cached pointer.
clearDatasetChunks/retireTopic/flushTo/replaceDatasetFrom/
enforceRetention (eviction erases chunks) free or move chunk memory — these
stay GUI-thread-only and a consumer must drop cached TopicChunk* (the
drop-before-mutate drain) before they run.DataEngine, and flushTo/
replaceDatasetFrom move chunks between the two. Code touching both mutexes
acquires them in ONE std::lock (the write host via lockEngineDeferred() ×2;
flushTo/replaceDatasetFrom already do). Never hold one engine's lock while
plain-locking the other — that ABBA-deadlocks against flushTo.Public compound mutators keep a locking wrapper + a private *Locked worker (e.g.
createTopic/createTopicLocked, commitChunks/commitChunksLocked) so an
internal call reuses the primitive without a redundant re-lock; with the recursive
mutex this is an efficiency detail, not a deadlock guard.
| Operation | Thread | Lock |
|---|---|---|
commitChunks (append) | worker | lockEngine(); never frees/moves chunks → no drain |
write host ensureTopic/ensureField/append*/flushPending | worker | lockEngine() per call (dual std::lock with staging when mirroring) |
createTopic/createTopicField | worker + GUI | lockEngine() (rehash; addresses stay stable) |
read cursors (rangeQuery/series/forEach) | GUI | lockEngine(), held for the view's lifetime |
latestAt | GUI | lockEngine(); value materialized before release |
getTopicStorage (raw ptr) | any | none — caller must hold lockEngine() for the pointer's lifetime |
clearDatasetChunks/retireTopic/flushTo/replaceDatasetFrom/enforceRetention | GUI only | lockEngine() + drop-cached-pointers drain first |
DerivedEngine recompute / addSisoTransform/addMimoTransform | GUI | lockEngine() across the read-modify-write |
The ObjectStore half has its own independent std::shared_mutex (see
docs/OBJECT_STORE_DESIGN.md); the two stores' locks are independent and must
never nest. Plugin sources that receive data on network threads still queue
internally and process in onPoll().
Pause/resume WriteCore swap (host-side, outside the engine lock). On pause,
DatastoreSourceWriteHost::setTarget / DatastoreParserWriteHost::setTarget
rebuild the host's WriteCore on the GUI thread while the ingest worker may still
be inside the old one — a data race + use-after-free on the WriteCore object /
state_->core, not on engine state, so lockEngine() cannot cover it (both racers
touch the same WriteCore). It is made safe by holding state_->core as an atomic
shared_ptr — AtomicSharedPtr<WriteCore>, a small mutex-backed stand-in for
std::atomic<std::shared_ptr<WriteCore>> (libstdc++ ships that specialization only
from GCC 12, and the build/CI baseline is GCC 11): each worker callback load()s a strong
reference that pins the core for the call's duration, and setTarget builds the
replacement fully, then publishes it with a release store(). An in-flight append
finishes on the old core (kept alive by the worker's reference) — the same accepted
semantics as the object-store host's std::atomic<ObjectStore*>. Regression-guarded
by engine_thread_safety_test's SetTargetSwapVsWorkerEnsureTopicRace under
ThreadSanitizer (./build.sh --tsan, and the Linux CI tsan job).
24 core test executables covering all layers (the authoritative live set is the
PJ_DATASTORE_TESTS list plus the explicitly-added targets in CMakeLists.txt):
| Test | Coverage |
|---|---|
buffer_test | RawBuffer, BitVector |
column_buffer_test | TypedColumnBuffer per-type append/read, validity |
type_registry_test | Schema registration, lookup, evolution |
encoding_test | All 5 encoding types: constant, FOR, dictionary, packed bool, raw |
chunk_test | TopicChunkBuilder row/bulk paths, seal, read-back |
topic_storage_test | Commit ordering, eviction, metadata |
out_of_order_ingest_test | out-of-order ingest (overlapping chunks, retention floor) |
query_test | RangeCursor, latestAt, edge cases |
series_reader_test | SeriesReader, SeriesCursor, series sample bounds/lookups |
engine_integration_test | Full DataEngine + DataWriter + DataReader round-trip |
derived_engine_test | SISO/MIMO transforms, topological order, incremental + batch recompute |
array_expansion_test | expandArray, clamping, cross-builder expansion |
regression_test | Bug-specific regression cases |
sequential_uid_test | SequentialUID allocation |
object_store_test | ObjectStore owned/lazy push, latest-at, retention, flushTo |
plugin_data_host_object_test | Object-write host bridges (pj.source_object_write / pj.parser_object_write) onto ObjectStore |
plugin_data_host_object_read_test | Toolbox object-read host (pj.toolbox_object_read) queries |
plugin_parser_object_write_test | Parser plugin writing canonical builtin objects through the object-write host |
streaming_mirror_test | streaming two-engine flushTo/setTarget mirror |
sample_test | Sample value type |
data_processor_test | PJ::proc::DataProcessor base |
processor_siso_adapter_test | ProcessorSisoAdapter (processor as DerivedEngine node) |
arrow_import_test | Arrow IPC schema parsing and batch import |
arrow_stream_round_trip_test | v4 Arrow C Data Interface round-trip (Phase 1b) |
Disabled pending the Phase 1b v4-ABI rewrite (commented out in CMakeLists.txt,
.cpp files still on disk): plugin_host_write_test (v3 appendArrowIpc/readSeries
write path) and plugin_host_read_test (v3 toolbox read path; rewrite for
read_series_arrow).
Build and run from the PJ4 repo root: ./build.sh, then ctest --test-dir build (e.g. ctest --test-dir build -R object_store_test).