Introducing Reladomo - Enterprise Open Source Java ORM, Batteries Included! (Part 2)

In part one of our investigation of Reladomo ORM, we explored usability and programmability features, as well as some of the core values that guided its development. In this second and final part, we'll have a look at the performance, testability and enterprise-focused features of Reladomo.

Performance

High performance solutions are a cornerstone of large, scalable enterprise applications. From a framework perspective, performance has two aspects:

1. The framework code itself should be well optimized.
2. The patterns exposed by the framework should allow the application code to implement high performance code.

When it comes to ACID database interaction, the most critical concern boils down to doing IO correctly. In Reladomo, we also recognize that latency is by far the bigger problem compared to bandwidth, so we optimize around that. Simply put, "doing IO correctly", means writing code that minimizes, combines (batch) and spreads (multi-thread) IO.

Minimize IO

We've already mentioned the deepFetch and advanced relationship facilities in Reladomo that significantly reduce IO over object graphs, in the first article. In the read-path, Reladomo's fully custom cache can have a dramatic effect on IO reduction, and we'll cover more of that below. In the write-path, multiple writes to the same object are automatically combined in the same unit of work, minimizing the calls to the database.

Combine IO

Two features that allow applications to properly combine their IO when doing queries are: 

• Reladomo's support for temp objects that map to temporary tables, and
• its support for tuple sets in queries.

Consider the Balance object in our ledger example. If we want to retrieve Balances from a pair of Account/Product lists, we can say:

MithraArrayTupleTupleSet tupleSet = new MithraArrayTupleTupleSet();
tupleSet.add(1234, 777); // Add account and product to the query
tupleSet.add(5678, 200);
tupleSet.add(1111, 250);

TupleAttribute acctAndProd = BalanceFinder.acctId().tupleWith(BalanceFinder.productId());

Operation op = BalanceFinder.businessDate().eq(today);
op = op.and(BalanceFinder.desk().eq("A"));
op = op.and(acctAndProd.in(tupleSet));

BalanceList balances = BalanceFinder.findMany(op);

For small sets, Reladomo translates this to an or-and clause. For larger sets, a temporary table is used (behind the scenes) to join to the Balance table.

In the write-path, facilities for bulk operations, including automatic batching, help consolidate write operations. Reladomo will reorder writes in a transaction to allow maximum batching without compromising correctness. Reladomo will also choose a batching strategy that’s appropriate to the size of the work and the database. For example, Reladomo can choose between four different insert strategies (bulk, union, multi-value, jdbc-batch).

Distributing IO

There are two main models Reladomo employs that help with spreading IO.

1. Sharding, as described below. Sharding allows more simultaneous writes because more hardware can be used and the shards don't need to worry about locking as much.
2. Reladomo's contract for object identity (only one persistent object for a given primary key in the entire JVM) makes it simple to reason about multi-threaded code. Writing multi-threaded code can have dramatic effects on IO. Many of the base APIs in Reladomo expose built-in multi-threading, such as MithraList.setNumberOfParallelThreads and utilities such as MatcherThread and MultiThreadedBatchProcessor.

Reladomo's multi-threaded loader is a good example of a generic utility that puts all these concepts together. It covers a simple, but highly recurring and reusable use-case: given a large incoming dataset (e.g. a file), what is the most efficient way to insert/update/delete the corresponding rows in the database? This use case is typical when data is copied from one system to another in a batch setting. Using the multi-threaded loader, the source and destination are read asynchronously, compared and efficiently written back. The components involved are the MatcherThread, which does the comparisons, the InputThread and DbLoadThread that read the source and sink and the SingleQueueExecutor that does intelligent batching of the writes. The use of many threads for reading, comparison and writing is how we spread the IO. The DbLoadThread streams the database data via Reladomo's forEachWithCursor method, which minimizes the reads. The SingleQueueExecutor is intelligently combining IO to reduce deadlocks. It takes only a few lines of code to wire up an instance of the multi-threaded loader. If you have this use case, you should give it a spin!

Application design is critical to taking advantage of the facilities provided above. In our ledger example, here are some things that significantly boost throughput:

• Sharding is used to spread the IO.
• In each shard, multiple trades are processed in the same unit of work to minimize commits, minimize queries and combine writes.
  • For example, when looking up products for incoming trades, we take a bunch of trades and do a single query to the database to find all of the products.
• Balance calculation writes to the database in batches.
• UI queries to multiple shards happen in multiple threads.

Reladomo's Cache

Reladomo comes with a custom cache that's much better suited to the needs of an ORM than any generic cache could be. The Reladomo cache is not a map. It's a collection of keyless indices, where each index is either a searchable set or multi-set. The attributes that form the identifiers of a particular index are arbitrary. The cache always has a primary key index and other indices are added based on application declarations or relationships between objects.

Reladomo guarantees that an object with a given primary key exists just once in the JVM and that allows the cache to cover the entire JVM. This makes the cache much more useful than a session based cache. It also makes it a lot more efficient than having a secondary serialized cache because there is no need for double storage of the same object and no need for serialization/deserialization. Unlike other ORM caches, the objects in the Reladomo cache are the objects visible to and used by the application.

The cache is also fully aware of Reladomo's transactional context. Changes (insert/update/delete) in a transaction are visible in the transaction, but only become visible outside after the commit.

The cache can be configured to be populated on demand (based on the queries being fired) or fully at startup. A full cache can be very useful for small static objects (e.g. country or currency). Under the right circumstances, a full cache can also be useful for larger datasets.

The cache structures are also fully customized for temporal object storage (see below). Temporal objects require very different in-memory storage because the business primary key does not uniquely identify a row. Reladomo's SemiUniqueDatedIndex is a singular data structure that hashes the same data in two different ways.

A more technical discussion of the cache can be found here.

Cache Notification

In an enterprise setting, it's unlikely that just a single JVM will access or change a particular dataset. This poses a serious issue for ORM caches. Reladomo's cache supports inter-cache notification to solve this problem. The notification is constructed as a set of lightweight cache-expiration messages that are delivered over a broadcast network. Out of the box, Reladomo comes with a TCP implementation of the broadcast network, suitable for a few hundred JVMs. The TCP implementation is a simple hub-spoke model, but includes dual hubs for failover. It's also quite easy to plug in other broadcast implementations by implementing a couple of small interfaces. The notification is described in more detail here.

Replicated Off-Heap Cache

Depending on application access patterns, some problems are best solved with an in-memory architecture. If the data lives in a database (and maybe it's sharded and bitemporal), inflating and keeping the in-memory cache up to date can be difficult for several reasons:

• Keeping data in sync with the database requires a strategy that doesn't leave the cache in an indeterminate state for a long time.
• Java GC can struggle when the heap size reaches 100GB and the application creates garbage during computations.
• Off-heap cache structures can cause even more garbage problems if objects are serialized in and out of the off-heap area.
• Off-heap cache structures have a difficult time maintaining relationships between objects.
• When the needs of the computation outgrow a single node, the cache should ideally be available on multiple nodes.

Reladomo's off-heap cache addresses all these issues.

• Reladomo leverages the audit temporal dimension for keeping the cache in sync with the database and replicas. The application code is always guaranteed to see consistent data up to a known milestone, which advances as new data comes in.
• The cache structures enable zero-deserialization access to all the data, which is very important for applications that address large portions of the whole (e.g. for aggregation). This also helps with keeping GC to a minimum.
• The object API for off-heap cached objects is exactly the same as on-heap objects, so the application code does not need to change based on the caching mode.
• The same index structure available to the on-heap cache is available to the off-heap cache, with the index data also kept off-heap.
• Caches of several hundred GB perform quite well.
• Caches are replicated using an efficient algorithm. Having ten replicas served from a single master is straightforward.
• Just like on-heap objects, relationships between objects are resolved dynamically. This allows the off-heap cache to have a simple, normalized structure.
• Strings are handled specially. More technical details can be found in these slides.

Enterprise Features

Code and data in an enterprise setting require a broader consideration than a traditional ORM provides. An enterprise grade ORM needs to cater to the full life cycle of the objects, from inception to retirement.

Consider the textbook example of a securities ledger with the following requirements:

• Fast trade processing for incoming trades.
• Balance computation based on the incoming data (trades, prices, etc).
• Heavy user-interactions that include time-based queries:
  • How have these balances changed over the last day, week, month?
  • What are the set of trades that affected this account in the last month?
  • What's the profit and loss for fixed income products in the last quarter?
• Ability for users (or upstream data) to correct past events, without losing reproducibility.
• Purging of data that is no longer required for reporting.

To fully realize all of these in a unified codebase, Reladomo provides a broad set of capabilities.

Sharding: Horizontally Scalable ACID

ACID (Atomic Consistent Isolated and Durable) is one of the fundamental storage assumptions in Reladomo. Scaling ACID requires deliberate design, and Reladomo provides built-in sharding to support that. The shard identity is stored with the object in memory (but not on the persistent store) and becomes part of the object's full identity. That allows simple construction of traditional primary keys without worrying about global uniqueness. For example, a trade in shard A can be given a simple integer id "17" which is reused in other shards for other trades.

Sharding is handled as a first class part of the Reladomo API. That means interacting with a sharded object does not require configuration-switching or code-segregation. A single query can span many shards and it's a natural part of the Finder API, treating the shard attribute like any other:

Operation op = BalanceFinder.businessDate().eq(today);
op = op.and(BalanceFinder.desk().in(newSetWith("A", "DR", "T", "ZZ")));
op = op.and(moreOperations());

BalanceList balances = BalanceFinder.findMany(op);
balances.setNumberOfParallelThreads(3);

The above will hit all the listed shards, in multiple threads (if outside a transaction).

The ledger example can make use of this facility by using a sharded design. Reladomo leaves that sharding strategy decision up to the application. Generally, one of two strategies is used:

1. Random sharding based on (universal) hashing.
2. Business-aligned sharding, based on some business concern.

The ledger trade processing pipeline can look like this:

Trades come in from various upstream sources. They are routed to the appropriate shard and processed (usually in a FIFO fashion) in that shard. Operations on different shards can be arranged to be completely independent. This architecture can scale well into hundreds of shards, easily handling millions of trades.

Bitemporal

Core requirements for a ledger include the ability to reproduce previous reports and attribute activity in the correct temporal order. By reproducible, we mean the ability to get the exact same results for queries done in the past. By "correct temporal order", we mean we can fix the view of the past events with new information, without affecting the results of queries done in the past. Take the simple case of a trade that hits the ledger a day later than it should have. First, any action taken because of this new information cannot change the result of old queries. If a user asks: "We sent a report of the trading activity for yesterday at 5 pm yesterday. What did it look like?" the ledger system must be able to answer that correctly. Second, the ledger must be able to put this trade into yesterday's flow of events. So when a user asks: "Given everything we know now, what does trading activity for yesterday look like?" the ledger system must be able to show the new trade in that data. There are two time-dimensions in this problem. One dimension is for full reproducibility. In literature that's called "transaction time" and we call it "processingDate" in Reladomo. The other time-dimension is for representing the business's view of events, regardless of real-time occurrence. In literature that's called "validity time" and we call it "businessDate" in Reladomo. Processing date is completely handled by Reladomo and it always corresponds to wall-clock time. Business date is entirely flexible and the application has to decide how to handle it.

From an API perspective, temporal concerns are baked into every aspect of Reladomo's implementation, by including four fields, as we will see below. The time dimensions are part of the query API:

Timestamp today = toTimestamp("2017-05-03");
Timestamp lastEvening = toTimestamp("2017-05-03 18:30");
Operation op = BalanceFinder.businessDate().eq(today);
op = op.and(BalanceFinder.processingDate().eq(lastEvening));
op = op.and(BalanceFinder.desk().eq("A"));
op = op.and(moreOperations());
BalanceList balances = BalanceFinder.findMany(op);

This will retrieve balances for the particular business date and processing date. The object domain API includes getBusinessDate() and getProcessingDate(). In other words, every object used in the application is located at some point in the two dimensional temporal space. The values are stamped on the object at retrieval time or during construction.

The ability to declare temporal relationships between objects is one of the highlights of the Reladomo framework. If we retrieve a balance object and ask for its account or product, those objects will correspond to the same point in temporal space. The entire navigable object graph represents a consistent temporal view of the data. This notion is carried over to queries as well: when a relationship is navigated in a query, the temporal information is applied to the relationship in the query.

Timestamp thirdQuarter = toTimestamp("2016-09-30");
Timestamp lastYear = toTimestamp("2016-12-01 20:00");
Operation op = BalanceFinder.businessDate().eq(thirdQuarter);
op = op.and(BalanceFinder.processingDate().eq(lastYear));
op = op.and(BalanceFinder.desk().eq("A"));
op = op.and(BalanceFinder.product().productName().startsWith("S"));
BalanceList balances = BalanceFinder.findMany(op);

In the above query, productName corresponds to what it was last year.

Writing to bitemporal objects requires a wider set of APIs and conceals a considerably more complex implementation. The APIs cover the following:

• Querying as of a point in 2D time.
• Navigating the object graph in a consistent temporal manner.
• Querying for history.
• Inserting and terminating as of a particular point in business time.
• Modifying as of a particular point in business time, optionally until another time.
• Incrementing a numeric field as of a particular point in business time, optionally until another time.

The API was developed from the needs of real-world applications that were already using bitemporal storage. For a more thorough walkthrough of the API, have a look at the bitemporal chapter of the Reladomo Kata. Let's look at an example. If we store the quantity balance for an account based on trading activity, after two trades, the balance can look like this in the database:

If we then add a trade back in time, with the following code we can update the balance:

Timestamp oldTradeDate = toTimestamp("2005-01-12");
Operation op = BalanceFinder.businessDate().eq(oldTradeDate);
op = op.and(BalanceFinder.desk().eq("A"));
op = op.and(BalanceFinder.balanceId().eq(1234));
Balance balance = BalanceFinder.findOne(op);

balance.incrementValue(40);

The effect of this on the balance data in the database is this:

Reladomo performed two inserts and two updates from that one line code incrementValue. The rules governing modification of bitemporal data is reasonably complex. Expecting every developer to implement them on top of a traditional ORM would lead to bugs and potential data loss. It's this behind the scenes implementation that is the value proposition in Reladomo's bitemporal API.

Rounding It Out

There is a collection of smaller features in Reladomo that make working in an enterprise setting smoother. Some of these include:

• Timezone support: mark fields for auto-conversion to UTC or the database timezone.
• Temp objects: use the equivalent of temp tables in your Java code.
• Mutable primary keys: for composite primary keys, optionally mark some of the parts as mutable.
• Metadata facilities: write code more generically using provided interfaces and runtime metadata. Many of Reladomo's built-in utilities do just that.
• 3-tier deployment: need to constrain your total connections to the database or cache some data? Setup a 3-tier setup, where one JVM acts as a hub for other JVMs.
• Purging and archiving API allow retiring or moving of data that is no longer required.

Testability out of the box

As every good developer knows, every application that's meant to run in production should have well tested code. That's doubly true for enterprises that have high regulatory scrutiny, reputational risk or security concerns (such as finance and health care).

Testing code that interacts with a persistent data store presents several problems:

• Setup and teardown of test data can be cumbersome.
  • It's tempting to setup incomplete objects graphs for a specific test, which easily leads to the test breaking when the production code starts using other parts of the graph.
• Handwriting mocks to represent interactions with the data store is problematic.
  • A simple test, where the production code performs writes followed by reads, would require some sort of mock state management.
  • Developers can't see the generated SQL or easily reason about the IO.
• Traditional setups where the tests run against a real database are hard to manage for a large team and put extra burden on each developer.

Reladomo's built-in testing framework solves these issues by providing a test resource that instantiates an in-memory database for a full sample, see Reladomo Test Resource. The database is populated from a set of easy to manage text files. There is no need to code up hard to read and modify insert statements. All interactions can be tested with no physical resource: the tests can run on a machine with no network connectivity. Developers can easily inspect the generated SQL (no question marks!) and reason about IO, for example, for spotting a missing deepFetch or optimizing writes. Full life cycle testing where the mix of all operations (read and write) is completely supported.

Reladomo's testing framework has been part of the code since the first release. It's used extensively for testing Reladomo itself.

One of the best uses of the built-in test framework is for uplifting legacy code. Introducing tests for existing code that's not well tested and using Reladomo to re-implement it allows for a high confidence replacement of the legacy code.

Conclusion

Testability, performance and enterprise features of Reladomo make writing object oriented code for ACID database interaction a reality. Reladomo fills the gaps in traditional ORM features in the enterprise space. By providing the APIs to write better and less code and test that code seamlessly, developers can spend their time on the high level design features of their application and their business logic.

If you're interested in learning more about Reladomo, you can try the Kata exercises, (our set of Reladomo tutorials) which incidentally, use the Reladomo test framework, so you don't need a database installed to try it out. Feel free to visit us on Github and have a look at the documentation.

About the Author

Mohammad Rezaei is a Technology Fellow in the Platforms business unit at Goldman Sachs. Mohammad is the chief architect of Reladomo. He has extensive experience writing high performance Java code in a variety of environments, from partitioned/concurrent transactional systems to large memory systems that require lock free algorithms for maximum throughput. Mohammad holds a B.S. in Computer Science from University of Pennsylvania and a Physics PhD from Cornell University.