FWD » Core Development » Database

A Bloom filter is a space-efficient probabilistic data structure used to quickly test whether an item has been seen before.

• If any bit is 0, the record is definitely new.
• If all bits are 1, the record is probably already seen.

4. New records are added by setting the corresponding bits to 1.

• Very memory efficient.
• Extremely fast lookups and inserts.
• Scales well for large datasets.

The memory consumption is reasonable, for a 50 million records table it is:
1% false positive rate → ~57 MB and ~7 hash functions
0.1% false positive rate → ~86 MB and ~10 hash functions

I think implementing one filter per table is most effective, this way we can configure the memory usage per table, as not all of them are equally used for insertions.

The filters will be persisted and kept on disk to ensure data integrity between server startups. They will be initialized once, before server startup, this means reading all the records from the whole database, or at least the tables that will be configured to use the in-memory validation.

I expect this approach to massively reduce the database chattiness for validation, which becomes a source of general slowness when latency is taken into consideration.

Ovidiu, the new approach is much simpler than that. The initialized filters hold only bits, some of them are set based on the data in the database, on initialization all records are read from the db, this can be done after db import for example. On record insert or update, the filters are further populated. There is no record caching, the stored information servers only one purpose, to answer if a record, or rather a set of fields is already present in the table. Uncommitted data checks stay the same, I am looking to reduce the validation by query only.

on initialization all records are read from the db,

This is a red flag for me. We are talking about tens of millions of record. This may require some tens of GB of memory to gather.

Alexandru Lungu wrote:

This is a red flag for me. We are talking about tens of millions of record. This may require some tens of GB of memory to gather.

The data is highly compressed, as I mentioned, for a 50 million records table, the filter uses 50mb of memory or more based on the desired efficiency.

Sorry to play the devil's advocate.

I understand that the data structure is some kind of SQL index, mapping from the precomputed hashes to PKs in database. But to compute the hashes, they need to be fetched from SQL to FWD at some time, before they are actually needed. This is not feasible for large tables, as Alex mentioned. Unless the hashes are computed on SQL, at import time and the server picks those on initialisation. In these conditions, what do we do when the database is updated externally?

The structure is a bit array, a very large one, 5 million for example. On write the indexed fields are hashed a few times and compute a few places (bits) in the bit array to be set, on read the same hash functions are applied and the bits are checked, if the bits are all unset, the record 100% does not exist in the db, if some of the bits are set, the record might exist, there could be some overlap between rows depending on the entropy of the hashes. If the record might exist, we do a database check like we used to.

If the database is externally updated, the filters must be cleared and repopulated, but I wonder if there are other FWD internal structures and mechanism that might be affected by an external update. I assume it is not advisable to insert in the db through means other than FWD.

This is a largely used data structure in production applications that deal with big data, mainly used for validations of new user data like username or email, I think our use case is quite similar.

I am planning to execute the filters initialization as Ant task, it would need to be executed after the import, strictly before the first server startup. This task would read the table names and a minimal config from the directory for which the filters are required, populate the filters and persist them on the disk. On server startup the artifacts will be loaded in memory and used (and maintained) based on the application execution (search/inserts).

I need a way to specify what tables need to use the filters and what config to apply to them, since filter population is a separate process, we can't use directory. I think the cleanest way is to use a separate file with this format:

filter.connection.url=jdbc:postgresql://localhost:5432/fwddb
filter.connection.username=fwd_user
filter.connection.password=user

filter.bloom.enabled=true
filter.bloom.tiers=small,medium,large

filter.bloom.tier.small.expectedInsertions=100000
filter.bloom.tier.small.falsePositiveProbability=0.005
filter.bloom.tier.small.tables=country,state,warehouse

filter.bloom.tier.medium.expectedInsertions=1000000
filter.bloom.tier.medium.falsePositiveProbability=0.005
filter.bloom.tier.medium.tables=customer,vendor,item

filter.bloom.tier.large.expectedInsertions=10000000
filter.bloom.tier.large.falsePositiveProbability=0.005
filter.bloom.tier.large.tables=order,invoice,transaction

Three different tiers for a different expected insertion frequency, higher tiers consume more memory, the tables that are not specified will not use the filters.
For multi-tenant applications, tenant API will be used to extract the connection data and the same config will be applied for all tenant databases.

I just realized that tenant API needs a running server, I think as a first option, manually writing all the database connection data will suffice.

Artur Școlnic wrote:

I just realized that tenant API needs a running server, I think as a first option, manually writing all the database connection data will suffice.

To be clear: you mean we will back this configuration needs described in #11526-13 with a REST API call to a running FWD server, instead of a configuration file?

If so, good.

We will not add external configuration files to the system.

I actually intended to use an external file. Adding tenant API for filter management is the better and more elegant solution, but this adds complexity to the overall task, additionally this means that the customer will have to start the server, use tenant API to configure the filters, populate the filters with an Ant task and only then will they be available for use, if this is acceptable, I will continue with this approach in mind.

I don't understand why this would be associated with tenant data. This is a general solution for persistence, right?

Is the configuration relatively static? Then it should live in the directory. A REST API can provide access if needed.

Why is the loading process separated from the rest of our code? We already have a database import process that accesses the directory and this is just an extension of that import process.

Greg Shah wrote:

I don't understand why this would be associated with tenant data. This is a general solution for persistence, right?

I could implement this as a general solution, this would mean that all the tables in all databases would have a filter, not only is this memory intensive, it is not necessary. I am sure there are tables that would benefit from a filter more than others and some that are rarely inserted into. This creates a need for per-table configuration and, in a multi tenant environment, per database config.

Is the configuration relatively static? Then it should live in the directory. A REST API can provide access if needed.

If we can use the directory without a running server, then yes, this could be done, although I would prefer the config to be stored in the database as it could contain hundreds of tables.

Why is the loading process separated from the rest of our code? We already have a database import process that accesses the directory and this is just an extension of that import process.

This can be done alongside the import, but this means that every time the filters need to be rebuilt, an import needs to be performed as well, I think having it as a separate process is cleaner. The filter population should not be a slow operation, it is just reading all the data from the db and calculating hashes, for additional performance this is done concurrently with a large number of threads. Once I have a working implementation, I will report on the time it takes with a customer db.

My point is that this should be a general solution not something hard coded to multi-tenant.

Yes, you can read the directory outside of the running server. We have classes that can be used for that. The import does that.

It can be written as code that is called at import time and also can be called separately. Most customers will just use it at import and then rely upon the ongoing server process to maintain it.

Whatever the process, how are you planning to persist the results? We really really don't like creating extra filesystem resources that have to be managed, secured, maintained... it sounds to me like you were planning a separate data store for persistence.

The easiest and most performant way is to store the artifacts (serialized filters) on the server filesystem, for distributed systems like stateless clustering there are 2 approaches, either build and store the filters on a single node and share it with the others, or have each node build it's own from the same database state. What we want to avoid it filter rebuild on server startup, this is not how persisted filters are meant to be used for obvious performance reasons.

When it comes to security, the filters have no readable data, it is hashed and highly compressed. The more concerning aspect is if someone writes to it, then we could treat a record as unique when it is in fact not unique. I will give this more thought, at this moment it will be as save as other FWD related files like directory.xml.

I tested the filter population on a customer db, I used 10 tables with 7.2 mil rows in total, the filters took only 15 seconds to initialize. TBF this is in a local env, so no latency penalty, still the selects are simple, fetch in large batches and only fetch a few fields that are part of an index, also all processing is done on primitive data, there is no added ORM processing time.

This is a level of speed that means we don't need to persist things in between server restarts. We could just build this on the fly at startup.

For stateless clustering, we would need to sync the edits.

Greg Shah wrote:

This is a level of speed that means we don't need to persist things in between server restarts. We could just build this on the fly at startup.

We don't know how much time it would take in a cloud env with 100 databases with tables having tens of millions of rows. The standard approach is to persist the filters on the disk.

For stateless clustering, we would need to sync the edits.

Yes, or keep copies on each node.

As far as memory consumption for the same 10 tables with 7mil rows goes, these are the values
if all where in the small tier (100k expected inserts): 1.4MB
if all where in the medium tier (1mil expected inserts): 14MB
if all where in the large tier (10mil expected inserts): 132MB

As a reminder, the larger the filter, the more effective it is. This is why we need a per table configuration, we can't have them all to consume 10MB of memory. On a positive note, the memory consumption is not dynamic and is known prior to committing to using it, so we will not have unexpected memory explosion. I will think about an algorithm that loads the filters on demand and maybe limits the nr of loaded filters, maybe a LRU cache would do the trick.

For now, make the location of the cache configurable. Production deployments must carefully place and secure all files. Adding this persistent cache makes deployments harder. If we find that we can initialize the cache without persisting it, we will move that way.

This is a level of speed that means we don't need to persist things in between server restarts. We could just build this on the fly at startup.

We can do this per-table asynchronously. In other words, the first X seconds of the server, a separate thread would build the filter. To assess memory consumption, we can consider limiting the number of tables for which we do this as a LRU, but with a huge skepticism of invalidation due to the cost of rebuilding it.

• if a filter does not exist or is under construction, then query the DB.
• if a filter is under construction, then any update to that table should "queue" the update to the construction phase.
  • when a construction phase ends, it processes the updates in the queue. While updates are processed, new updates may be pushed, but this is fine IMHO.
  • we need to capture a critical execution block. When the queue is empty, the thread shall synchronize, wait for ongoing updates, block for ongoing updates, confirm the filter is built, continue normally.
  • I say all this to prevent blocking session in order to wait for filter building, but at the same time have the filter make consistent progress and prepare a minimal downtime.
• once a filter exists, it is used.
• updates on a table where a filter exists shall update the filter.
  • these single-point updates are blocking.

In the scenario of the OG task, we can allow (some) VALIDATE to query the database directly while the filter is built asynchronously. When the filter is ready, simply switch with a minimal critical code segment to the filter. This way, all other VALIDATE will be faster.

• rollback will be a PITA. A single transaction may do lots of updates and creates (some in sub-transactions). A rollback would revert the database state to a prior version. FWD doesn't care much about it, it simply runs ROLLBACK TO SAVEPOINT. There are some in-memory undoables that are snapshot (DMO data mostly). In some cases, we don't do that and it bites us back functionaly (e.g. updates in dirty-share database). In trunk, RecordNursery and DMO states are not rollbacked (fixed in #8388). This new filters proposed should also be sensitive to rollbacks. Would this means we will snapshot the touched filters together with SAVEPOINTS and we revert them to the saved snapshots back on rollback? If so, is memory enough to hold the filters + snapshots of filters for each nested sub-transaction?
• this is less of an issue, but it touches a fragile aspect: transaction isolation. Two different transactions are actively seeing different data (due to creates/updates/deletes done by that transaction). While UniqueTracker is ensuring a consistent view over the database across transactions, can we rely on this to ensure we do not need "per-session" filters.
  • Example for phantom read: what if a transaction creates a new record; this record can't be seen by other sessions (e.g. lack of cross-session dirty-share), but I think it is appended to UniqueTracker. So trying to make a new records or update a record to conflict will be rules out by unique tracker. I don't think this is a problem per se.
  • Example for phantom read or non-repeatable read: what if a transaction updates/deletes a record; this record can still be found by other sessions, but I think the LockManager will lock exclusively that row and the other session will pend anyway.

So, LockManager and UniqueTracker may ensure that filters should simply track the committed data, right? This need to be tested very hard.
Also, while filters are per committed data, we still need to handle un-commited data. How do we manage this?

• for a running FWD server, when does this gets persisted on disk? Can it recover in case the FWD server crashes and the filters didn't sync to disk? This is more about the possibility of having false negatives in the filter, thus the record is an unique index violation.
• for a clustering mode: when spinning up a new node, what is persisted on disk is obsolete. How can these be re-synced?

I like the non-persisted nature of the design in #11526-27.

I don't think we can apply changes to the filter before the transaction ends. Is there a way to use the filter in a stacked approach where there a session level set of changes that kept separate until commit?

Greg Shah wrote:

I like the non-persisted nature of the design in #11526-27.

I like it too, unless the population of the filter eats half the server resources for a long time and the other work done on it suffers dramatically, we would need to test both approaches. The persistent filters also allow more control and awareness over the memory consumption.

Is there a way to use the filter in a stacked approach where there a session level set of changes that kept separate until commit?

Yes, it does make sense to have the filters hold only committed data and have a more dynamic structure tied to the session that can be rolled back or committed at the end of a transaction. In this case the validation order is:
1. Check the uncommitted data via a new structure like a set of hashes compatible with the filter or an existing FWD functionality, if one exists (session cache, UniqueTracker).
2. Check the filter.
3. Check the database.

Constantin Asofiei wrote:

Artur, some questions:

• for a running FWD server, when does this gets persisted on disk? Can it recover in case the FWD server crashes and the filters didn't sync to disk?

For the ultimate correctness, the filters need to be persisted on commit, however this could prove costly, I was thinking that we can mark the filters dirty and flush them periodically, but if the server crashes, some data could have not reached the persisted filter, in this case all of them would need rebuilding.

• for a clustering mode: when spinning up a new node, what is persisted on disk is obsolete. How can these be re-synced?

For clustering mode, the filters would need to be synced from a single master node, or rebuild on startup individually. I am not sure how other FWD structures are synced now, but this one should not be any different.

So there will be a "buildup" of invalid (out of date) bits for deleted records? That is even more of a reason to re-initialize at server startup. We may even want a REST API to force a re-initialization in a long running server.

Greg Shah wrote:

So there will be a "buildup" of invalid (out of date) bits for deleted records?

Yes, but a sparse array will mitigate this issue, additionally, this only means that sometimes a query will be emitted because of a deleted record, but in theory this should occur very rarely. Of course completely in memory management would be preferred, but only if it proves to a relatively lightweight operation.

I'm surprised that the solution doesn't just clear the bits associated with the deleted record.

When a record is added, we hash its indexed fields and set the corresponding bits to 1. Those bits may already be shared by other records (not all of them, but a subset), so you cannot safely set them back to 0 on delete without risking false negatives.

Review of 11526a, r16607

• the serialised stream also contains the class names. This is not mandatory bad, but may affect (dilute) the entropy. For example, an Integer value is expected to be encoded as 4 bytes, but the implementation will add the bytes of java.lang.Integer and the variable length string representation of the value;
• as long the toString() returns full information of the value. This is not the case (because Record.data[] only stores Java native values) but in case of BDTs this would return values truncated by their (default) format.

I think it will work for prototyping, but for the final, optimized version, each data type should be binary encoded here. To note that there are a fixed set of these data types and, except for the data[] itself, there are no arrays, therefore a recursive implementation is not necessary.

II. Usage of the computed hash. I missed the following issue in my previous review:

long[] hash = BloomFilterHashUtils.hash(rowData);
// ...
filter.merge(new EnhancedDoubleHasher(hash[0], hash[1]));

• degradation - there are cases where a physical table is used as a temp-table - data is added for i.e. a report and after that delete. Such scenarios exists, may not be common. But we should measure a miss ration, if DB is hit more than x% of time, rebuild the filter. Is DB is again hit more than x% of time, discard the filter? More, at runtime, updates and deletes will degrade the filter, as it is built only from inserts.
• consistency - always rebuilt on server startup, while feeding records from the database table and FWD runtime. Only when the database table completes it will be used by runtime (same applies when rebuilding the filter).
• how much time does it take to build a filter with 100's of millions of records?
• what is the performance impact on maintaining the index filter? There may be unique indexes with many columns - if is less than the DB hit, it may be OK. But we need to understand the impact.
• what is the memory impact of the bloom filter? for clustering mode, there should be only one instance of it. We have a manual way of specifying which tables have this enabled, but a 40MB per filter times 1000 indexes is 40GB...
• does 128 bits provide enough space to not have 'false negatives' where DB is hit when it should not?
• the filter must be rebuilt at each FWD server startup. We can do this in parallel (see previous point about 'degradation'), and activate the filter only when it completed. But we will get the same issue as with dynamic queries, a fresh server will seem to work slow.

If you have an early implementation, we need some serious stress-testing, for 100s of millions of records, where multiple clients create/update/delete and measure the degradation (i.e. DB hit for unique validation). Provide both enough entropy via i.e. UUIDs and something like 'a', 'aa', 'aaa' (not random IDs).

Constantin Asofiei wrote:

• degradation - there are cases where a physical table is used as a temp-table - data is added for i.e. a report and after that delete. Such scenarios exists, may not be common. But we should measure a miss ration, if DB is hit more than x% of time, rebuild the filter. Is DB is again hit more than x% of time, discard the filter?

Yes, makes sense to discard an inefficient filter.

• consistency - always rebuilt on server startup, while feeding records from the database table and FWD runtime. Only when the database table completes it will be used by runtime (same applies when rebuilding the filter).

We need to test that on a large database, the typical use case for these filters is to persist them for some of the tables and rebuild the filters only when needed and the system is not heavily loaded, nightly for example.

• how much time does it take to build a filter with 100's of millions of records?

for 10 tables with 7mil records it takes 15 sec, we can extrapolate from that.

• what is the performance impact on maintaining the index filter?

minimal, the primitive data is fetched from the db and hashed, nothing more.

• what is the memory impact of the bloom filter? for clustering mode, there should be only one instance of it. We have a manual way of specifying which tables have this enabled, but a 40MB per filter times 1000 indexes is 40GB...

Please see note 25.

• does 128 bits provide enough space to not have 'false negatives' where DB is hit when it should not?

yes, 128 bits is enough for the hasher input here. The DB-hit rate is governed by the Bloom filter sizing config, not by false negatives from the 128-bit hash.

Artur Școlnic wrote:

• consistency - always rebuilt on server startup, while feeding records from the database table and FWD runtime. Only when the database table completes it will be used by runtime (same applies when rebuilding the filter).

We need to test that on a large database, the typical use case for these filters is to persist them for some of the tables and rebuild the filters only when needed and the system is not heavily loaded, nightly for example.

A persisted filter and the FWD server down will miss any direct changes to this table. So the filter will no longer be consistent... I don't think we can re-use a persisted filter, we always need to recreate.

Ok, I was reluctant to go with this approach since it would cripple the server on startup for an undetermined amount of time, but the arguments for it keep adding up, I will change the implementation to be fully in memory.

Artur, rebuild the index at FWD server must be done concurrently (the runtime will work without the filter, and the filter becomes active when it completes to build from the database). Also see the degradation point, this same mechanism should be used to rebuild it if needed.

• the FWD server starts normally
• the filter build starts in the background (in a few threads?)
• the filter can accept notifications for events from the runtime, but the runtime does not use it until it completely reads the table
• when the table has built the filters for all indexes, these filters are seen by the runtime for reading also.

Review of 11526a, r16609

• FilterManager.java:
  • Some methods quickly return if the table/database are temporary. This is correct, but I wonder what should we do with the other 'special' databases: dirty and meta. They do not seem to be excluded and the filters seem to apply to them as well. To note that some tables of meta database are indeed stored with the permanent tables and the one which may benefit from this optimisation would be the _User. There are customers which do write to this table and checking the uniqueness might be improved.
  • the initializeTableFilter() relies on columnOffsets map to locate the offsets of a specific column. But we know each index offset component. What I what to say is that we may drop the slow map access and replace it with table access from a bitmap (as we do for indexes). But this optimisation my require a bit more of work so it can be delayed after proving the correctness of the solution.
• BloomFilterHashUtils.java
  • line 159: I still think this is not correct: notice the parameters: EnhancedDoubleHasher(long initial, long increment). But the passed argument are the low and high 64 bytes of the hash returned by MurmurHash3.hash128().

Ovidiu Maxiniuc wrote:

• BloomFilterHashUtils.java
  • line 159: I still think this is not correct: notice the parameters: EnhancedDoubleHasher(long initial, long increment). But the passed argument are the low and high 64 bytes of the hash returned by MurmurHash3.hash128().

EnhancedDoubleHasher uses initial and increment as the two base hash values in enhanced double hashing: initial is h1(x), and increment is h2(x); MurmurHash3.hash128() returns a 128-bit digest split into two 64-bit longs, so passing hash⁰ and hash¹ supplies those two base hashes directly.

I addressed the other concerns, Ovidiu, please review rev 16611.

Performance for the #11477 scenario (with added 0.5ms latency) is

baseline
cold: 20.3s
warm: 18.7s

11526
cold:17.8s
warm:16.2s

Alexandru, could you please test POC performance, if I recall correctly it contains item and supplier record writes. I don't expect an improvement in that scenario, just want to make sure that it does not regress it.

Smoke testing and stress testing of a customer app with 11526a did not reveal any regressions.

We need to move this ahead. If this is disabled by default, is it safe to merge?

The current implementation is not suited for a multi tenant env, the memory consumption will be too high. I need to implement a limiting initialization, maybe to a set budget of memory or number of tables. For a single tenant project the current implementation could work fine, depends on the database. It is hard to estimate the impact on memory without knowing the exact data that will be used.

I am ok with presenting this feature as a prototype for now, which is easily disabled, but certainly not as a finished product that will perform well in any circumstances (not yet).

For memory management I am implementing 3 things:
1. Making filter sizing table-specific instead of one fixed medium size for everything.
2. Hook database/tenant disconnect cleanup into the existing lifecycle
3. Add a global cap, by bytes and evict least-recently-used filters, the global budget is configurable via directory.

Review of 11526a/r16612

• FilterManager.java:
  • line 10: drop 003 counter. Only need a single entry per commit;
  • lines 73 & 75 should be dropped (compacted import statement block);
  • line 393: use diamond notation <> (drop compile-time inferred FlushedUniqueKeyOverlay);
  • line 451: result of filterLru.get(filterKey); is not used in any way. Maybe rewrite it as

    cached = filterLru.get(filterKey);

  • line 1005: missing empty line to separate @param from @return;
  • lines 1192, 1371, 1384: methods can be static;
  • line 1209: both calls to this method use false for 1st parameter (from lines 1016 & 1074);
  • line 1386: javadoc mentions "if present". In that case, slash is -1 and substring will fail with StringIndexOutOfBoundsException;
  • lines 1081, 1091: these can be merged into a single line:

    CachedFilter oldLru = filterLru.put(filterKey, cached);

    similar to line 1080. Also, I think cachedBytes -= old.bytes should be enough. Let me know if I am wrong;
  • line 1096: I think cachedNow will always be true here, since filterKey was added to filters at line 1080. If evictUntilWithinBudget() will evict it using LRU, then the cache is empty?
  • line 1156: evictedFilters.remove(filterKey); should be preferred. We know that retryAt is its value;
  • line 1173: this is a no-op. evictedFilters can contain filterKey if evictionCooldownMs > 0. This could have happened only if evictionCooldownMs was changed meanwhile;
  • lines 1261-1287: Although this is a rarely called method, it is potentially costly because it iterates all those maps looking for prefixed keys. In similar situations I used 2 level maps. The first level being the database and the second the table. This will replace the current flat key (db-name/table). The exact performance gain (or loss) remains to be obtained through testing, if you choose to go this way;
  • lines 1481-1533: this method can be extracted in some utility class and made available for all FWD code;

Ovidiu, thank you for the feedback, please take a look at rev 16613.