PStates

There are two ways in which schemas are implemented by Rama from an indexing perspective. Either the entire value is indexed, or each element of the collection is indexed separately. Top-level mapSchema or fixedKeysSchema index elements separately, and marking a nested schema as "subindexed" will cause that nested data structure to index its elements separately. All other schemas / class references index as single values.

PState schemas specify subindexing like so:

"$$p" contains a subindexed set inside of a map, "$$p2" contains a subindexed map inside of a map, and "$$p3" contains a subindexed list inside a non-subindexed map inside of a map. You can also have subindexed structures inside other subindexed structures – this functionality composes infinitely.

The granularity of indexing determines how much work is done for reads and writes. If a structure is not subindexed, the entire structure needs to be read from disk even if you’re just querying for one element of it. For example, consider this query on "$$p3":

Suppose the contents of "$$p3" on that partition are:

The query navigates into the partition, step by step. The first step key("a") operates on the top-level map which indexes its elements separately, so it will read the value for "a". Because that submap in the schema is not sub-indexed, that value stores the entire contents of the map. So key("a") will read from disk:

If the data structure being resolved only has a small number of elements, like less than 50, the performance difference from being subindexed or not will be insignificant. But if it has a large number of elements, like 10 million, there will be a massive performance difference.

The next step key("b") reads the value for "b". Because that list is subindexed, it actually only reads from the already fetched map a unique identifier for the subindexed structure called the "reference ID". So instead of reading from disk the entire list, which could contains millions of elements, it only reads a small 8 byte value per subindexed structure. The data structure at that point of navigation acts just like a normal list, but to resolve list queries it uses its internal reference ID to find the individual elements on disk.

The next step nth(1) does a lookup on that subindexed list, which reads the value 11 from disk. The values 10 and 12 as well as all the other values in the other subindexed lists are never read from disk in this query.

Subindexed maps and sets are sorted. You can do efficient range queries on them using the navigators sortedMapRange, sortedMapRangeFrom, sortedMapRangeTo, sortedSetRange, sortedSetRangeFrom, and sortedSetRangeTo. These navigators are extremely smart about how they iterate on data on disk to resolve the substructure, generally doing just a single disk seek and a scan to resolve the entire substructure. For an example of usage, see this section from the Paths page.

PStates have a few options available when declaring them. PStates use the builder pattern so you can easily string multiple options together.

The global option causes the PState to have a single partition on task 0 rather than a partition on every task. It’s used like so:

An example use case for a global PState would be storing the "top 10" of some entity being tracked. It’s also commonly used for things like global counts and sums.

The option initialValue specifies the value to initialize each PState partition to when the module is first launched. It’s used like so:

initialValue can only be used if the PState schema is a class reference. It’s commonly used with global, but there’s also many use cases for non-global PStates that aren’t top-level maps. One example is the PState used for ID generation in ModuleUniqueIdPState from rama-helpers.

The option makePrivate makes the PState inaccessible for reads for anything other than the owning topology. Here’s an example of usage:

Trying to access a private PState via RamaClusterManager or from another module will result in an exception.

Finally, the last PState option available is keyPartitioner. This changes how PState clients determine to which PState partition to send a query. For example, consider the following PState client query:

The query extracts the key "a" from the query path and uses the configured key partitioner to route the query to the right partition. The default key partitioner is hash-based, and you can use the keyPartitioner option to implement things like sorted partitioning schemes.

The key partitioner is a function taking in as arguments the number of partitions for the PState and the extracted key. It returns the partition number to send the query. Here’s an example of configuring a key partitioner that routes every request to the last partition:

The configured key partitioner is also used when calling select in topologies, described more in the next section.

localSelect queries the PState partition located on the current task. It emits one time for every navigated value. For example:

This topology code queries for every element inside a nested collection within "$$p". If the collection for a key has zero elements or doesn’t exist, the localSelect call won’t emit at all. If it has ten elements, localSelect will emit ten times.

localSelect is a synchronous call. Nothing else can happen on a task while a localSelect is running, meaning no other PStates on that task can change. You can string together multiple localSelect calls to look at values from multiple PStates synchronized to a moment in time.

select works just like localSelect except partitions the computation first based on the query path. For example:

The path given to select must start with key, and the key provided will be used with the configured key partitioner on the PState to choose the target task. The key partitioner defaults to a hash-based partitioner and is equivalent to using hashPartition. Also, select always emits on the task it partitions to and not on the task it started from (unless those happen to be the same).

localSelect and select callsites accept an option "allow yield" which is useful for queries that need to operate over a large amount of data from the PState. For example, you may want to emit every element of a subindexed set that could have hundreds of thousands of entries in it. The "allow yield" option allows queries like this to be done without holding the task thread for an extended amount of time. See this section for a discussion of why it’s important to not hold the task thread for more than a few milliseconds at a time.

When the "allow yield" option is specified, the localSelect or select call will execute over the course of many individual events – potentially thousands of them. Here is an example of specifying the option:

The results of the query are identical to as if that option were not specified. The only change is how it executes. Even though the query executes over many events, it operates over a stable view of the PState. So any transforms that happen to the PState after the query begins will not be seen by the query.

It’s important to note that when that option is set, subsequent code may not be synchronous with the start of the query. When allowYield is specified, multiple PStates cannot be guaranteed to be read from at the same point in time like you can when that option is not specified.

Yielding selects automatically break up large chunks of work with pagination, such as when navigating to every element of a subindexed structure with .all(). The default page size is specified with the dynamic option pstate.yielding.select.page.size, and it can be adjusted mid-path with the .withPageSize navigator. For example:

If the default page size is set to 250, the first .all() and the .mapKeys() will use a page size of 250, while the second .all() and the .mapVals() will use a page size of 100. Different page sizes may be more optimal for data of different sizes, with experimentation needed to determine the best setting for any particular use case. Rama’s default performs well for most use cases.

PStates are updated by using localTransform, agg, or compoundAgg. As mentioned earlier, they can only be written to by their owning topology. Stream and microbatch topologies implement PState writes very differently in order to achieve high performance.

localTransform uses transform paths to specify updates. Just like queries, transform paths use navigation to target subvalues within a PState. See the documentation on the Paths page for details.

A higher level mechanism for updating PStates is by using aggregators through agg or compoundAgg. See the Aggregators page for details.

Lastly, querying PStates from other modules is almost exactly the same. You can use localSelect and select in the same way, with only a minor difference in how you tell Rama which partition of the target PState to interact with. Since queries on mirror PStates execute asynchronously anyway, they always use yielding selects. Specifying the .allowYield() option manually is thereby not allowed.

See the Module dependencies page for details.

When using select or selectOne, the PState client must decide to which PState partition to send the query request. If the PState has only one partition, by being declared global or the module having only one task, then the query is always sent to that partition and there are no restrictions on the path used for the query.

When the PState has more than one partition, the select and selectOne variants that only take in a path as input (like shown in the previous section) use the path to determine which PState partition to query. They require the path begin with the key navigator, and the key from that navigator is extracted and used to choose the PState partition. This key is given to the "key partitioner" configured during the PState declaration to make that choice. The key partitioner defaults to a hash-based partitioner if not configured, which is equivalent to using hashPartition in a topology.

There’s another variant of select and selectOne which take in the partitioning key as an additional argument. This variant uses this explicit partitioning key along with the configured key partitioner to route the query, and there are no restrictions on the path used for the query.

An example of where an explicit partitioning key is useful comes from our Twitter-scale Mastodon implementation. Mastodon identifies "statuses" (Mastodon’s name for tweets) by a combination of "user ID" and "status ID". The "Core" module in Mastodon handles both user profiles and status indexing. Three of the PStates in "Core" are:

All these PStates are kept colocated by the user ID of a status. Colocating them greatly improves the efficiency of rendering timelines for users, which involves fetching all associated profile information and status stats (like count, boost count, etc.) for a collection of statuses. However, the "$$statusIdToFavoriters" PState doesn’t include the user ID as part of its PState structure because it’s unnecessary and inefficient. Status IDs are already unique within the module. So the "$$statusIdToFavoriters" PState is partitioned by a value (the user ID) that isn’t part of the data within it.

The explicit partitioning key variants of select and selectOne allow the "$$statusIdToFavoriters" PState to be queried by a PState client like so:

This queries the number of likes for a particular status. The user ID is used as the partitioning key, and the status ID is used to navigate the PState partition. More generally, the explicit partitioning key variants of select and selectOne are useful whenever a PState is partitioned by something other than top-level keys in its structure.

The select and selectOne variants shown so far will block the thread while waiting for a response from the cluster. There are also non-blocking variants of these that return CompletableFuture objects.

Here are a few examples of using the non-blocking API:

CompletableFuture has a rich API for composing with other CompletableFuture objects and for registering callbacks for notification of success or failure.

The diff types in Rama represent all the different ways a map, list, set, or multi-set can change. A map can have keys updated, values updated, or entries removed. A list can have elements appended, inserted, updated, reordered, or removed. Sets and multi-sets can have elements added or removed.

All these diffs are broadly categorized as top-level diffs or low-level diffs. A top-level diff captures change as compactly as possible, while a low-level diff specifies change in terms of locations like keys or indexes. Many diffs can be converted into one or more other diffs during processing. For example, a SequenceInsertDiff represents a single element being inserted in between two indexes in a list and contains two fields: an index and a value. Since inserting an element in the middle of a list grows the list by one element and changes the value for every index after the insertion point (by shifting every value by one index), one SequenceInsertDiff can be converted into a collection of many KeyDiff and one AppendDiff.

Diffs sent as callbacks from a PState to a ProxyState are always top-level diffs. The full set of diffs available in Rama are shown in the diagram below, with arrows indicating how diffs convert to other diff types:

The diffs colored in blue are top-level diffs. The diffs colored in green can also be top-level diffs, but have special meaning:

The Javadoc for each diff type explains what the diff means, what fields it has, and how it converts to other diffs. You’ll see in the next section how you can choose which diff types you wish to process and how Rama will automatically do any diff conversion necessary.

The diffs produced on a PState are determined entirely by the paths used in localTransform callsites or the aggregators used in agg or compoundAgg callsites. The diffs are computed incrementally as part of the transform, and they’re only computed if there are any subscribers on the PState. Not every transform affects every subscriber, and only a portion of a diff may be applicable to any particular subscriber. Let’s take a look at an example of this:

The module in this example produces a global PState with a map of counts. The main method creates proxies on the top-level of the PState and to the values for the "a" and "b" keys. Running main prints:

Here you can see the diffs produced at the top-level of the PState are always KeyDiff, but the proxies for "a" and "b" only receive the portion of that diff that affects their values. Underneath the hood, PStates separate subscriptions by key to efficiently skip them when the transform doesn’t affect them.

The top-level diffs are KeyDiff because that’s the diff produced by the key navigator in this case. If key were used in conjunction with termVoid, it would produce a KeyRemoveDiff instead. The Javadoc for each navigator lists the diffs they produce.

agg and compoundAgg callsites can also be used to update PStates. Underneath the hood these callsites compile to localTransform calls using paths. CompoundAgg.map compiles to key and CompoundAgg.list compiles to multiple calls to nth.

The specificity of produced diffs depends on the specificity of the corresponding transforms. If a transform precisely targets a subvalue, the diff will be very specific. However, if it does something coarse-grained like a termVal on an entire substructure just to change one value nested within, the diff will be coarse-grained and inefficient to process by subscribers. Fortunately, highly specific paths are easier and more natural to use anyway.

Another note about this example: the top-level schema is Map.class instead of PState.mapSchema(String.class, Integer.class) because you cannot proxy objects which index their elements separately (top-level map or subindexed schemas). You can proxy values within them, just not the object itself.

Capturing diffs in proxy callbacks and analyzing them is the basis for powering higher level reactivity in an application. For example, your diff processing could trigger fine-grained changes to your frontend UI.

Diffs contain a double dispatch based API for processing. This API makes it easy to specify which diffs you want to process while having Rama automatically convert any diffs for you. Let’s take a look at an example:

Diffs are processed by calling process on them with an object implementing the processing logic. Every diff type defines an inner interface Processor with a method specific to that diff. KeyDiff.Processor defines processKeyDiff, AppendDiff.Processor defines processAppendDiff, and so on. To specify the diff types in the hierarchy you wish to handle, your processor just implements the Processor interfaces for those diff types. In addition, the processor object must implement Diff.Processor which defines the method unhandled.

After calling process, Rama will expand that diff until every expanded diff can be handled by the provided processor. If that’s impossible, no processing is done and unhandled is called instead.

Running this prints:

In the first case, you can see the processor only implements KeyDiff.Processor but was able to process a KeysDiff. If you consult the diagram earlier on this page, you’ll see this is because KeysDiff expands to many KeyDiff. In the second case, the provided processor cannot handle UnknownDiff so the unhandled method is called and the result is set to null.

process will process diffs in the order in which they happened. This can be significant for a case like a SequenceInsertDiff and a SequenceChangeDiff, where the order in which those happened can completely change the result. For the above example, each key in a KeysDiff is independent so the order in which each key’s diff happened doesn’t affect the result. So KeysDiff is expanded in an arbitrary order.

Although the diffs produced for a proxy are entirely predictable from the paths used in the PState transforms, it’s critical to still handle the case where unhandled is called on your processor. During normal operation, the diffs sent back to a ProxyState will be the fine-grained diffs expected from the transform paths. However, if something goes wrong – like a network error or overloaded cluster – a ProxyState can be forced to resync from scratch. In these cases it will re-execute the path and produce a ResyncDiff. Your processing code in this case should analyze the new value in its entirety and perhaps do a comparison against the old value. If you are using reactivity to render a UI, this could involve re-rendering an entire component from scratch.

As long as your cluster is appropriately provisioned, these recompute scenarios should be extremely rare. The vast majority of the time you should expect your reactivity code to receive fine-grained diffs. But for an application to be completely robust, it’s important to handle these rare failure scenarios as well.

proxy has an extremely robust implementation, carefully designed to ensure the value inside ProxyState matches the value that would be produced from freshly executing the path used to create the ProxyState.

Internally, diffs are sent back to a ProxyState along with a CRC. The CRC is checked after applying the diff, and if it doesn’t match the ProxyState will resync from scratch. This ensures integrity of the ProxyState in the face of any of the numerous failures that could happen in a distributed system.

A ProxyState also heartbeats to its subscribed PState partition on a fixed interval. This continuously validates it’s still subscribed on the PState partition, and it also lets the server know the ProxyState is still active so as not to cleanup the subscription. If the heartbeat finds the subscription is not there, the ProxyState will resync from scratch.

An active ProxyState uses resources on the process of the subscribed PState partition. The PState partition uses memory to track the subscription and uses CPU to process diffs for the subscription. The PState client for an active ProxyState also uses resources to regularly heartbeat and process incoming diffs.

All state for a ProxyState on both the client and server can be cleaned up by calling close on it. It’s best to call close as soon as a ProxyState is no longer needed in order to clean up resources as promptly as possible. Calling close also sends a final DestroyedDiff to the callback for the associated proxy call.

A PState partition will still clean up stale subscriptions that weren’t closed properly. If a PState partition hasn’t received a heartbeat from a ProxyState after a configurable timeout, it will clean up all resources used for the subscription and stop considering it for diff processing.

Rama provides protection against a bad proxy path (that causes an exception when being executed) repeatedly failing. By default, if a proxy errors 10 times within 120 seconds, it is forcibly terminated and an UngracefulTerminationDiff is sent to its callback if specified. This is true whether the error happened on the initial query for the proxy or the reactive processing to handle changes. The failure window size and threshold can be configured.

As discussed above in the section on subindexing, a location in a PState is either indexed as a whole or has its elements indexed individually. The only parts of a PState that can have migrations declared with PState.migrated are full values. You can’t declare a migration on a location that’s indexed as part of a bigger value. The only locations that can have migrations are values of top-level maps or values of subindexed structures.

For example, here’s a valid migration:

Since the containing map is subindexed, the value of that map can be migrated. On the other hand, this is not a valid migration:

Since the containing map is not subindexed, the targeted value is part of a greater value and a migration cannot be performed. That migration could instead be written like this by moving the migration up to the complete value:

Likewise, since locations with a fixed keys schema are indexed as a single value, you cannot declare a migration on a value in a fixed keys schema. This is invalid:

Instead, you’d have to declare the migration on the fixed keys schema as a whole.

Finally, keys of top-level or subindexed maps and values of subindexed sets cannot be migrated. So a migration like this is invalid:

If you try to deploy an invalid migration to a Rama cluster, the deploy will be rejected.

If you need to do a migration that’s not supported by the migration API, the best approach is usually to recompute a new PState from depot data.

There can be any number of migrations on the same PState, including nested within one another. Here’s an example:

In this example you can see the fixed keys schema is being migrated by removing the "aliens" key, and the values of a subindexed map nested within are being migrated as well. Migration IDs are tracked per location and are used to determine what to do when a module is updated while it is still migrating values on disk, as described in the next section.

You don’t need to wait for a migration to finish to update a module. If you leave the PState definition the same and don’t change any migration IDs, the migration will continue where it left off after the module update. However, if you change any migration IDs the entire migration will restart for the entire PState.

You might want to restart a migration for various reasons – you had a bug in your migration code, you want to tweak how you’re migrating those values, or you want to additionally migrate other parts of the PState.

For a restarted migration, some of the values seen by the migration function will have been migrated the first time around and some will not have been migrated yet. If you restart multiple times, there could be even more kinds of values. A restarted migration function must idempotently handle any of the kinds of values that could be seen.

Rama’s Cluster UI provides detailed information on migration progress. The main page and module page show if any migrations are still in-progress for any of their PStates:

The module instance page shows which PStates are in progress or done with migrations:

Diving into a PState page shows which partitions of that PState are still working on the migration:

Scrolling below that shows time-series telemetry on the migration progress:

The first chart gives an estimate on the total progress of the migration based on the number of top-level keys that have had everything below them fully migrated versus the total number of top-level keys. It’s an estimate rather than an exact calculation because Rama only knows the approximate number of keys in a top-level map. The second chart shows the number of top-level keys that are fully migrated over time. The last chart shows the number of migration paths that have been applied to the PState. There could be many migration paths per top-level key if there are multiple values being migrated under a top-level key (e.g. all the values of a subindexed map nested within).

When the Cluster UI shows a migration as complete for a PState, you’re free to remove the PState.migrated wrappers from the PState’s schema on the next module update. If you try to remove those before the migration is complete, Rama will reject the module update.

Migrations happen as part of topology execution. Before each streaming batch or microbatch completes, a limited amount of migration work is done on a PState. If multiple PStates have migrations specified, the PStates are migrated one at a time. When a migration completes on a PState after some number of batches, it moves on to the next PState.

Each round of migration work iterates over part of the PState and applies transform paths to change and replicate the values on disk. The amount of iteration done in each round can be controlled by dynamic options. There are separate dynamic options for streaming and microbatching because they have different defaults:

A migration uses additional task thread load, and the defaults increase task thread load by about 10% for common types of migrations. However, this can vary depending on the specifics of the migration, such as the cost of running the migration function.

If the module doesn’t have spare task thread load before the update, we recommend scaling the module first before deploying the migration.

Schema changes that are equivalent to or a generalization of that part of the PState schema can be done without declaring a formal migration on that location. These types of schema changes don’t cause any disk migration to be done in the background. For example, if the original PState schema is this:

You can do a module update with this schema without declaring a migration:

Likewise, if your original PState schema is this:

You can do an implicit migration like this:

On the other hand, trying to change that schema to this would be rejected:

When Rama only knows the original type was Map.class, it can’t be migrated to a PState.mapSchema with anything other than Object.class keys and values since Rama doesn’t know what types were stored in those locations.

You can also migrate a location from PState.mapSchema with any key/value schemas to Map.class, as that is a generalization.

These examples were just for maps, but the same holds true for equivalent or generalized schemas for lists and sets as well.

You can try out migrations using in-process cluster (IPC). You can simulate multiple versions of the same module by overriding the getModuleName() method in the module definition. In a real project, the different versions of a module would be separated in version control, but to play with any sort of module update you’ll need both versions on the same classpath.

For example, here’s one version of a module:

Now suppose you want to update it to this version of the module along with a migration to the PState:

The migration turns the number values in the PState into a string and appends a "!" to them. You can try out this migration in IPC like this:

This starts the first version of the module, appends some data, and then updates the module to the new version. It prints out the values of those keys before and after the update:

The "after update" values will print with the migration applied whether or not the migration has actually completed on disk yet.

Here are the configurations available for PStates. Configurations for modules are fixed upon module deploy and can only be changed with a module update. They cannot be customized on a topology by topology basis. Likewise, configurations for clients also can only be changed by creating a new RamaClusterManager.

Dynamic options can be edited from the Cluster UI and take effect immediately. They can be configured on a topology by topology basis. The dynamic options available for PStates are: