Skip to content

README.md

Latest commit

 

History

History
127 lines (98 loc) · 5.4 KB

README.md

File metadata and controls

127 lines (98 loc) · 5.4 KB
Raw

event-reduce (Python)

Python implementation of the EventReduce algorithm. It takes the previous results of a query and a change event of the database and calculates the new results in memory, without running the query over the database again.

It uses the BDD that the JavaScript implementation created, so both always decide on the same action. It needs Python 3.10 or newer and has no dependencies.

Installation

pip install event-reduce

Usage

from event_reduce import (
    ActionName,
    ChangeEvent,
    StateResolveFunctionInput,
    calculate_action_name,
    dict_query_params,
    run_action,
)


def compare(a, b):
    # must never return 0 for two different documents, so sort by the primary key last
    if a["age"] != b["age"]:
        return -1 if a["age"] < b["age"] else 1
    return -1 if a["id"] < b["id"] else 1


# SELECT * FROM users WHERE age > 18 ORDER BY age, id LIMIT 10
query_params = dict_query_params(
    primary_key="id",
    sort_fields=["age", "id"],
    query_matcher=lambda doc: doc["age"] > 18,
    sort_comparator=compare,
    limit=10,
)

results = [{"id": "a", "age": 20}, {"id": "b", "age": 30}]

# a change event from the change stream of your database
change_event = ChangeEvent.insert("c", {"id": "c", "age": 25})

action = calculate_action_name(StateResolveFunctionInput(query_params, change_event, results))
if action == ActionName.RUN_FULL_QUERY_AGAIN:
    results = run_query_on_database()  # EventReduce could not optimize this event
else:
    run_action(action, query_params, change_event, results)
    # -> [a, c, b]

dict_query_params() is for dict documents, like the primaryKey and sortFields of the JavaScript implementation. The sort fields can be dot-separated paths like "address.city". For other documents, create the QueryParams with a get_document_id function and a sort_fields_changed(previous, doc) function that returns true if a sort field changed.

Key document map

For big result sets, pass a dict from the id to the document as key_document_map to both StateResolveFunctionInput and run_action(). It is used to check whether a document is in the results without scanning them, and run_action() keeps it in sync with the results.

Differences to the JavaScript implementation

  • Change events are created with ChangeEvent.insert(), ChangeEvent.update() and ChangeEvent.delete().
  • The names are snake_case, ActionName and StateName are enums whose values are the JavaScript names.
  • run_action() raises an EventReduceError for RUN_FULL_QUERY_AGAIN and UNKNOWN_ACTION. ALWAYS_WRONG is only used by the truth-table generator and raises like UNKNOWN_ACTION.
  • calculate_action_name(), resolve_input() and ResultWindowBuffer() raise an EventReduceError with the code NEG_LIMIT for a negative limit. Like in JavaScript, a limit of 0 returns no documents and the action is then always DO_NOTHING.

Result window buffer

For a query with a limit or a skip, EventReduce returns RUN_FULL_QUERY_AGAIN when a row has to move into the result window that is not in the previous results: the row after the window (when a row leaves a full window) or the last skipped row (when a row is added in front of the window). The result window buffer avoids most of these full queries: it keeps limit_buffer_size rows after the window and skip_buffer_size skipped rows before it in memory. It works like the result window buffer of the JavaScript implementation, which explains how it works, how to choose the buffer sizes and has measurements.

from event_reduce import ResultWindowBuffer, get_recommended_result_window_buffer_size

# recommended size: about log2(limit), at most 30
size = get_recommended_result_window_buffer_size(query_params.limit)
buffer = ResultWindowBuffer(query_params, limit_buffer_size=size, skip_buffer_size=size)
# run the query with the skip and limit of the buffer
buffer.reset(run_query(buffer.query()))
# the rows of the original skip and limit
results = buffer.results()

# on new change events
result = buffer.apply_change_events(change_events)
if result.run_full_query_again:
    buffer.reset(run_query(buffer.query()))
    results = buffer.results()
elif result.changed:
    results = result.results

query() returns the skip and limit to run the query with, limit is None if all rows after the skip are needed. math.inf as a buffer size keeps all rows on that side, like Infinity in JavaScript.

Development

pip install pytest ruff mypy
pytest
ruff check . && ruff format --check . && mypy src

The BDD is generated by the truth-table generator of the JavaScript implementation, which also writes src/event_reduce/bdd_generated.py. The tests run the shared test suite, the same tests as all other implementations.

Who uses EventReduce

RxDB, the local-first database, uses EventReduce for its live queries since version 9.0.0, where it is enabled by default. When an app subscribes to a query, RxDB keeps the results up to date by running each write event through EventReduce, and only queries the storage again when EventReduce returns RUN_FULL_QUERY_AGAIN.