Data Management

The data module provides HDF5-backed storage, Cap’n’Proto serialization, and archive file handling.

Overview

FusionSC uses a statically typed, cross-version-compatible, high-speed binary message format for data storage and handling. It provides three persistence mechanisms:

Archive files: Store an entire tree of a root message and its dependencies in a single immutable file. Optimized for fast writing and extremely fast reading.
Structured IO: For import and export to other programs, supports JSON, YAML, CBOR, and BSON formats.
Warehouses: Mutable object stores backed by a local SQLite database, supporting multi-user access, compact storage, and backup facilities.

API Reference

namespace kj: KJ library (https;//capnproto.org)

namespace fsc

The FSC library.

Simple interface module for ZLib

Typedefs

template<typename T> using References = typename internal::References_<T>::Type: Use this to figure out what datatype a DataRef points to.

Functions

template<typename T> constexpr bool isTemporary()

template<typename T, typename Cap = capnp::FromClient<T>, typename ...Attachments> Cap::Client attachToClient(T src, Attachments&&... attachments)

template<typename T, typename = kj::EnableIf<capnp::kind<capnp::FromReader<T>>() == capnp::Kind::STRUCT>> bool hasMaximumOrdinal(T in, unsigned int maxOrdinal)

Struct version checker.

When passing structured data between functions, Cap’n’proto will silently hide all fields which are not understood by the current version of the protocol. Likewise, if new fields are added, methods that do not yet understand them will likely not check for their presence. In many cases, this is useful behavior.

However, for scientific codes, when data are requested, it is extremely important that a given request is understood completely. This method aims to enable protocol evolution with this contraint, by checking whether a function, that can only interpret fields up to the given ordinal number in the given struct, will be able to fully understand the passed data. It does so by not only inspecting the ordinal numbers of set fields, but also by inspecting the wire representation of the struct, to ensure that there are no unknown fields set by a potentially newer version of the protocol.

Parameters:

in – The data to be checked for consistency.
maxOrdinal – An upper bound on the ordinal number of fields. Any field with an ordinal number exceeding this parameter (or unknown to this client) may only be set to its default value, otherwise this method will return false.

Returns:

true if the given struct meets the max ordinal requirements, otherwise false.

Variables

constexpr capnp::ReaderOptions READ_UNLIMITED = {std::numeric_limits<uint64_t>::max(), std::numeric_limits<int>::max()}

template<typename T> struct _IsTemporary

Public Static Attributes

static constexpr bool val = false

template<typename T> struct _IsTemporary<fsc::Temporary<T>>

Public Static Attributes

static constexpr bool val = true

template<typename T = ::capnp::AnyPointer> struct DataRef

Cap’n’proto interface for data objects.

The DataRef template is a special capability recognized all throughout the FSC library. It represents a link to a data storage location (local or remote), associated with a unique ID, which can be downloaded to local storage and accessed there. Locally downloaded data are represented by the LocalDataRef class, which subclasses DataRef::Client.

The stored data is expected to be in one of the two following formats:

If T is any type except capnp::Data, it must correspond to a Cap’n’proto struct type. In this case, the DataRef holds a message with the corresponding root type, and a capability table that tracks any remote objects in use by this message (including other DataRef instances).
If T is DataRef::Data, then the corresponding message can either be a raw binary or a Cap’n’proto message with a capnp::Data object as its root (note that the library will default to raw storage because of the laxer size constraints)

Once obtained, DataRefs can be freely passed around as part of RPC calls or data published in other DataRef::Client instances. The fsc runtime will do all it possibly can to protect the integrity of a DataRef. In the absence of hardware failure, data referenced via DataRef objects can only go out of use once all referencing DataRef objects do so as well.

DataRefs only represent a link to locally or remotely stored data. To access the underlying data, they must be converted into LocalDataRef instances using LocalDataService::download() methods.

Note

This is a generated Capnproto interface class. For notes on how to deal with such a class, please consult the Capnproto C++ examples (https://capnproto.org) or the generated API documentation

interface DataRef (T) {
    interface Receiver {
        begin @0 (numBytes : UInt64) -> ();
        receive @1 (data : Data) -> stream;
        done @2 () -> ();
    }
    
    metaAndCapTable @0 () -> (metadata : DataRefMetadata, table : List(Capability));
    rawBytes @1 (start : UInt64, end : UInt64) -> (data : Data);
    transmit @2 (start : UInt64, end : UInt64, receiver : Receiver);
}

Template Parameters:: T – Type of the root message stored in the data ref.

struct Client : public virtual capnp::Capability::Client

Actual interface object class.

Subclassed by fsc::LocalDataRef< T >

Public Functions

template<typename T2 = ::capnp::AnyPointer> inline DataRef<T2>::Client asGeneric()

struct DataService

Remote interface to data service.

This interface serves as an access point to remotely download

Note

This is a generated Capnproto interface class. For notes on how to deal with such a class, please consult the Capnproto C++ examples (https://capnproto.org) or the generated API documentation

interface DataService @0xc6d48902ddb7e122 {
    store @0 [T] (id : Data, data : T, schema : AnyPointer) -> (ref : DataRef(T));
    # Upload a message to the remote data service and have it publish it
    
    clone @1 [T] (source : DataRef(T)) -> (ref : DataRef(T));
    # Have the remote data service download a DataRef and re-publish it
    
    hash @2 [T] (source : DataRef(T)) -> (hash : Data);
    # Have the remote service inspect the linked tree of refs and their content tables and
    # compute a hash based on the received information.
    
    cloneAllIntoMemory @3 [T] (source : DataRef(T)) -> (ref : DataRef(T));
    # Have the remote data service download a complete copy
}

template<typename T = capnp::AnyPointer> class LocalDataRef : public fsc::DataRef<T>::Client

Local version of DataRef::Client.

Data reference backed by local storage. In addition to the remote access functionality provided by the interface in capnp::DataRef, this class provides direct access to locally stored data. This class uses non-atomic reference counting for performance, so it can not be shared across threads. To share this to other threads, pass the DataRef::Client capability it inherits from via RPC and use that thread’s DataService to download it into a local reference. If this ref and the other DataServce share the same data store, the underlying data will not be copied, but shared between the references.

Public Functions

ArrayPtr<const byte> getRaw(): Provides direct access to the raw underlying byte array associated with this data reference.

Array<const byte> forkRaw(): Same as getRaw(), but provides an owning reference (that can be passed across threads)

T::Reader get(const capnp::ReaderOptions &options = READ_UNLIMITED): Provides a structured view of the underlying data. If T is capnp::Data, the returned reader will be identical to getRaw(). Otherwise, this will interpret the backing array as a CapNProto message with the given type at its root. Note that if T is not capnp::Data, and the backing array can not be interpreted as a CapNProto message, this method will fail.

template<typename T2 = capnp::AnyPointer> class LocalDataRef<T2> as(): Provides a new data reference sharing the underling buffer and capabilities, but having a different interpretation data type.

ArrayPtr<const byte> getID()

ArrayPtr<capnp::Capability::Client> getCapTable()

DataRefMetadata::Format::Reader getFormat()

DataRefMetadata::Reader getMetadata()

LocalDataRef(LocalDataRef<T> &other)

LocalDataRef(LocalDataRef<T> &&other)

LocalDataRef<T> &operator=(LocalDataRef<T> &other)

LocalDataRef<T> &operator=(LocalDataRef<T> &&other)

LocalDataRef<T> deepFork(): Creates a copy of this DataRef (sharing the data) that can be safely passed to other threads.

LocalDataRef() = delete

LocalDataRef(DataRef<capnp::AnyPointer>::Client capView, Own<internal::LocalDataRefImplV2> backend)

template<typename T2> LocalDataRef(LocalDataRef<T2> &other)

Public Members

Own<internal::LocalDataRefImplV2> backend

Friends

friend class LocalDataRef

class LocalDataService

Publishes and downloads data into LocalDataRef instances.

Main entry point for handling local and remote data references. Can be used to both create remotely-downloadable data references with its ‘publish’ methods and download (as in, create local copies of) remote references with its ‘download’ methods.

Download methods

template<typename Reference, typename T = References<Reference>> Promise<LocalDataRef<T>> download(Reference src, bool recursive = false)

Downloads remote DataRef::Client into LocalDataRef.

Downloads the data contained in the remote reference into the local backing store and links the remote capabilities into a local capability table.

Parameters:

src – The DataRef<T>::Client to download from (can also be a LocalDataRef<T>)
recursive – Whether to recursively download all referenced DataRef instances into the local data store as well. If true, all contained DataRef objects will point into local storage after the returned promise resolved, and are therefore guaranteed to instantly resolve in future download attempts.

Returns:

Promise to a local data ref instance which extends the interface by DataRef with direct access to the stored data.

template<typename Reference, typename T = References<Reference>> Promise<Maybe<LocalDataRef<T>>> downloadIfNotNull(Reference src, bool recursive = false)

Downloads remote DataRef::Client into LocalDataRef if it is not null.

Downloads the data contained in the remote reference into the local backing store and links the remote capabilities into a local capability table.

Parameters:

src – The DataRef<T>::Client to download from (can also be a LocalDataRef<T>)
recursive – Whether to recursively download all referenced DataRef instances into the local data store as well. If true, all contained DataRef objects will point into local storage after the returned promise resolved, and are therefore guaranteed to instantly resolve in future download attempts.

Returns:

Promise to a local data ref instance which extends the interface by DataRef with direct access to the stored data. The contained Maybe will evaluate to nullptr if the given DataRef is unset.

Publication methods

template<typename T = capnp::Data> LocalDataRef<T> publish(typename DataRefMetadata::Reader metaData, Array<const byte> backingArray, ArrayPtr<capnp::Capability::Client> capTable = nullptr)

Publishes binary data and capability table.

Creates a local data reference directly from a backing array and a capability table.

The interpretation of the backing array depends on the seleced data type. If the data type is capnp::Data, the array is interpreted as the raw data intended to be referenced. This is e.g. intended to be used for raw memory-mapped files. Currently, any other type leads to the backing array being interpreted as containing a CapNProto message (including its segment table) with a root of the specified data type (can also be capnp::AnyPointer, capnp::AnyList or capnp::AnyStruct or similar).

Parameters:

id – The global ID to publish this data ref under.
backingArray – The contents of the binary buffer. The LocalDataRef object will take ownership of this buffer.
capTable – A list of capabilities (usually the capability table of the passed Cap’n’Proto message) to be stored alongside the binary data.

Returns:

A LocalDataRef object which can be passed into all targets expecting abort DataRef::Client.

template<typename Reader, typename T = capnp::FromAny<Reader>> LocalDataRef<T> publish(Reader reader)

Publishes Cap’n’proto message.

Creates a local data reference by copying the contents of a capnproto reader. If the reader is of type capnp::Data, the byte array it points to will be copied verbatim into the backing buffer. Currently, for any other type this method will create a message containing a deepcopy copy of the data referenced by this reader and store it into the backing array (including the message’s segment table). Capabilities contained in the reader’s message will be added into the capability table hosted by the DataRef.

LocalDataRef<capnp::Data> publish(kj::ArrayPtr<const byte> bytes): Publish the contents of an array (copies and hashes the array)

LocalDataRef<capnp::Data> publish(kj::Array<const byte> bytes, kj::ArrayPtr<const kj::byte> hash = nullptr): Take ownership of array and publish it, potentially with precomputed hash.

Archiving methods

template<typename Ref, typename T = References<Ref>> Promise< void > writeArchive (Ref reference, const kj::File &out) KJ_WARN_UNUSED_RESULT

Write DataRef to an archive file.

Downloads the target data and all its transitive dependencies and writes them into an archive file. This file can then be shared with other customers to provide them a deep copy of the stored data.

template<typename T> LocalDataRef<T> publishArchive(const kj::ReadableFile &in, const capnp::ReaderOptions readerOpts = READ_UNLIMITED)

Publish contents of archive file as LocalDataRef.

Reads an archive file and publishes all data contained within. Returns a LocalDataRef to the root used when writing the archive.

template<typename T> LocalDataRef<T> publishArchive(kj::Array<const byte> in, const capnp::ReaderOptions readerOpts = READ_UNLIMITED)

template<typename T> LocalDataRef<T> publishConstant(kj::ArrayPtr<const byte> in)

LocalDataRef<capnp::Data> publishFile(const kj::ReadableFile &in, kj::ArrayPtr<const kj::byte> fileHash = nullptr, bool copy = false): Publish the raw contents of a file as a data ref via mmap or copy.

LocalDataRef<capnp::Data> publishFile(const kj::ReadableFile &in, bool copy = false): Shorthand for publishing without hash.

Flat representation

template<typename Client> Promise<kj::Array<kj::Array<const byte>>> downloadFlat(Client src)

template<typename T> Promise<kj::Array<kj::Array<const byte>>> downloadFlat(LocalDataRef<T> src)

template<typename T> LocalDataRef<T> publishFlat(kj::Array<kj::Array<const byte>> data)

Raw data files

Promise<void> downloadIntoFile(DataRef<capnp::Data>::Client, Own<const kj::File> &&out)

Limit configuration

void setLimits(Limits limits)

Public Types

using Nursery = kj::Vector<kj::Own<void>>

Public Functions

void setChunkDebugMode(): Reduces chunk size to 1kB and throws error if chunks can’t be mapped.

operator DataService::Client()

LocalDataService(const LibraryHandle &hdl): Constructs a new data service instance using the shared backing store.

LocalDataService(LocalDataService &other)

LocalDataService(LocalDataService &&other) = default

LocalDataService &operator=(LocalDataService &other)

LocalDataService &operator=(LocalDataService &&other) = default

LocalDataService() = delete

Private Functions

LocalDataService(internal::LocalDataServiceImpl &impl)

Private Members

Own<internal::LocalDataServiceImpl> impl

Friends

friend class LocalDataRef

struct Limits

Public Members

uint64_t maxRAMObjectSize = 500000000

Maybe<uint64_t> ramRemaining = nullptr

template<typename T> struct Temporary : public T::Builder

Combined MessageBuilder and Struct::Builder.

This class holds a locally owned MessageBuilder and specifies the type of the message’s root. It directly derives from T::Builder, so the root struct / list can be directly accessed.

Template Parameters:: T – Data type of the contained message.

Public Types

using Builds = T 

Public Functions

template<typename ...Params> inline Temporary(Params... params)

inline Temporary(typename T::Reader reader)

inline Temporary(typename T::Builder builder)

inline Temporary(Own<capnp::MessageBuilder> holder)

Temporary(Temporary<T>&&) = default

Temporary<T> &operator=(Temporary<T> &&other) = default

inline Temporary<T> &operator=(typename T::Reader other)

inline Temporary<T> &operator=(std::nullptr_t)

inline T::Builder asBuilder()

inline operator capnp::MessageBuilder&()

Public Members

Own<capnp::MessageBuilder> holder

namespace fsc

The FSC library.

Simple interface module for ZLib

struct Blob

Blob object.

Public Functions

virtual Own<Blob> addRef() = 0

virtual void incRef() = 0: Increases the in-database refcount.

virtual void decRef() = 0: Decreases the in-database refcount and deletes the blob if it hits 0.

virtual int64_t getRefcount() = 0: Reads the current in-database refcount.

virtual kj::Array<const byte> getHash() = 0: Reads the hash of the blob (or nullptr if the blob is under construction or deleted)

virtual int64_t getId() = 0: Returns the ID of the blob.

inline bool isFinished(): Checks whether incRef may be called on this blob.

virtual Own<BlobReader> open() = 0: Opens the blob for reading.

struct BlobBuilder : public kj::OutputStream

Construction helper for new Blobs. Inherits from kj::OutputStream.

This interface is responsible for managing the construction of a new Blob for the store. It can only be obtained through BlobStore::create(). Upon creation, a blob under construction is created with an implicit reference count of 1. The reference count may not be increased until the blob’s construction is finished.

The blob may be filled with data through the OutputStream interface. Once all data have been transferred, call the finish() method.

finish() will check for hash duplication.If the new blob’s hash is unique, that blob will be returned (and from now on its refcount can be increased). In case of a hash conflict, the new Blob will be deleted, the originally present blob with same hash will have its refcount increased by 1, and that blob will be returned.

This allows for interactions where the lifetime of the under-construction blob is attached to an external object during the construction phase, and the Blob can be deleted implicitly if that object gets deleted.

Public Functions

virtual bool tryConsume(kj::ArrayPtr<const byte> input) = 0

Alternative write API. This does not call the underlying database and only manipulates the compressor and the buffer. It can be called from other thread as long as no simul taneous other calls to the blob builder are made.

Returns true if the buffer was completely consumed. If it returns false, flush() must be called before calling tryConsume again. In this case, the next call to tryConsume MUST hold the same data as the previous call.

virtual void flush() = 0: Flushes buffer to database. UNLIKE tryConsume CAN NOT BE USED CROSS-THREAD.

virtual Own<Blob> finish() = 0: After write completion and hash key assignment, returns the finished blob (not neccessarily the same as the blob returned by getBlobUnderConstruction()

virtual Own<Blob> getBlobUnderConstruction() = 0

struct BlobReader : public kj::InputStream

Extended input stream interface that can outsource the compression to an external thread. Inherits from kj::InputStream.

Public Functions

virtual Promise<size_t> tryReadAsync(void *buf, size_t min, size_t max, const kj::Executor &decompressionThread) = 0

struct BlobStore

Blob storage interface.

Public Functions

virtual Own<BlobStore> addRef() = 0

virtual Maybe<Own<Blob>> find(kj::ArrayPtr<const byte> hash) = 0: Looks for a blob object in the database based on the given has key.

virtual Own<Blob> get(int64_t id) = 0: Returns a blob by its database ID.

virtual Own<BlobBuilder> create(size_t chunkSize) = 0: Allocates a new blob without hash key in the database to be written to.

namespace fsc

The FSC library.

Simple interface module for ZLib

struct DataStore

Public Functions

DataStore(fusionsc_DataStore *store)

~DataStore()

DataStore() = delete

DataStore(const DataStore&)

DataStore(DataStore&&)

DataStore &operator=(const DataStore&)

DataStore &operator=(DataStore&&)

fusionsc_DataStore *incRef()

fusionsc_DataStore *release()

StoreEntry publish(ArrayPtr<const byte> key, Array<const byte> data)

Maybe<StoreEntry> query(ArrayPtr<const byte> key)

void gc()

Private Members

fusionsc_DataStore *raw = nullptr

struct StoreEntry

Public Functions

StoreEntry(fusionsc_DataStoreEntry*)

~StoreEntry()

StoreEntry(const StoreEntry&)

StoreEntry(StoreEntry&&)

StoreEntry &operator=(const StoreEntry&)

StoreEntry &operator=(StoreEntry&&)

fusionsc_DataStoreEntry *incRef()

fusionsc_DataStoreEntry *release()

kj::ArrayPtr<const byte> asPtr()

kj::Array<const byte> asArray()

Private Members

fusionsc_DataStoreEntry *raw = nullptr

namespace fsc

The FSC library.

Simple interface module for ZLib

namespace db

Enums

enum class TransactionType : uint8_t

Values:

enumerator UNKNOWN

enumerator READ_WRITE

enumerator READ_ONLY

enumerator READ_ONLY_DEFER

Functions

KJ_DECLARE_NON_POLYMORPHIC(Connection::BaseTransactionHook)

struct Connection

Public Functions

virtual Own<Connection> addRef() = 0

virtual Own<Connection> fork(bool readOnly) = 0

virtual Own<PreparedStatementHook> prepareHook(kj::StringPtr sql) = 0

virtual bool inTransaction() = 0

inline PreparedStatement prepare(kj::StringPtr sql)

inline int64_t exec(kj::StringPtr sql)

virtual Own<TransactionHook> beginTransaction(TransactionType) = 0

inline bool isSqliteLike()

Checks whether connection behaves like an SQLite connection.

If true, the connection has the following traits:

Serialization failure happens latest at the first write statement of a transaction.
It is strongly advisable to batch small write transactions into larger ones.

If these traits are true, clients can and should merge multiple write statements into super-transactions that will be released before the event loop goes to idle. If serialization failure can occur in a delayed fasion, this is not safe to perform as earlier successful writes can be invalidated by the later failure.

Protected Functions

Own<TransactionHook> beginTransactionBase(kj::StringPtr beginStatement)

Implementation helper for standard SQL transactions.

Starts transaction with custom begin statement
Uses COMMIT / ROLLBACK to manage the top-level transaction
Uses SQL savepoints to manage nested transactions
Committing or rolling back the top-level transaction has the identical effect on all nested savepoints.

Protected Attributes

bool sqliteLike = false

Private Members

Maybe<Own<BaseTransactionHook>> activeTransaction

struct TransactionHook

Subclassed by fsc::db::Connection::BaseTransactionHook

Public Functions

virtual ~TransactionHook()

virtual void commit() = 0

virtual bool active() noexcept = 0

virtual void rollback() noexcept = 0

struct PreparedStatement

Prepared statement object.

Public Functions

PreparedStatement() = default

PreparedStatement(Own<PreparedStatementHook> &&hook)

~PreparedStatement()

PreparedStatement(PreparedStatement&&) = default

PreparedStatement &operator=(PreparedStatement&&) = default

template<typename P> void setParameter(size_t, P): Set a statement parameter.

template<typename ...Params> Query bind(Params... params): Assigns all parameters from arguments.

inline void reset(): Reset the statement.

template<typename ...Params> inline size_t operator()(Params... params): Execute the statement directly (bind + step + reset). Returns no. of rows modified.

template<typename ...Params> inline int64_t insert(Params... params): Execute the statement directly (bind + step + reset). Returns last inserted rowid.

Private Functions

template<typename ...Params, size_t... indices> void bindInternal(std::integer_sequence<size_t, indices...> pIndices, Params... params): Helper method for binding with known index sequence.

Private Members

Own<PreparedStatementHook> hook

struct Column

Column of a statement row.

Public Functions

inline Column(PreparedStatement &parent, size_t idx)

inline kj::ArrayPtr<const byte> asBlob()

inline double asDouble()

inline int64_t asInt64()

inline kj::StringPtr asText()

inline bool isNull()

inline operator kj::ArrayPtr<const byte>()

inline operator kj::StringPtr()

inline operator double()

inline operator int64_t()

inline kj::String name()

inline bool operator==(const Column &other)

inline bool operator!=(const Column &other)

inline Column &operator++()

inline Column *operator->()

inline Column &operator*()

Public Members

PreparedStatement &parent

size_t idx

struct Query

Query object returned by bind() that can step through result rows and access columns.

Public Functions

inline Query(PreparedStatement &p)

inline ~Query()

inline Column operator[](size_t idx)

bool step()

Public Members

PreparedStatement &parent

struct PreparedStatementHook

Backend implementation class for prepared statements.

Public Functions

virtual void reset() = 0

virtual bool step() = 0

virtual int64_t lastInsertedRowid() = 0

virtual size_t nRowsModified() = 0

virtual void setParameter(size_t, double) = 0

virtual void setParameter(size_t, int64_t) = 0

virtual void setParameter(size_t, kj::ArrayPtr<const byte>) = 0

virtual void setParameter(size_t, kj::StringPtr) = 0

virtual void setParameter(size_t, decltype(nullptr)) = 0

virtual double getDouble(size_t) = 0

virtual int64_t getInt64(size_t) = 0

virtual kj::ArrayPtr<const byte> getBlob(size_t) = 0

virtual kj::StringPtr getText(size_t) = 0

virtual bool isNull(size_t) = 0

virtual kj::String getColumnName(size_t) = 0

virtual size_t size() = 0

struct Transaction

Public Functions

Transaction(Connection &parent, TransactionType type = TransactionType::UNKNOWN)

~Transaction() noexcept(false)

inline void commit()

inline void rollback() noexcept

inline bool active() noexcept

Private Members

Own<Connection::TransactionHook> hook

kj::UnwindDetector ud