rx-connector

Reindexer is an embeddable, in-memory, document-oriented database with a high-level Query builder interface. Rx-connector allows to connect to a Reindexer instance from java-application.

Maven

<dependency>
<groupId>com.github.restream</groupId>
<artifactId>rx-connector</artifactId>
<version>[LATEST_VERSION]</version>
</dependency>

Usage

Here is example of basic rx-connector usage:

//Define an item classpublicclassItem {
// 'id' is a primary key@Reindex(name = "id", isPrimaryKey = true)
privateIntegerid;
// add index by 'name' field@Reindex(name = "name")
privateStringname;
// add index articles by 'articles' array@Reindex(name = "articles")
privateList<Integer> articles;
// add sortable index by 'year' field@Reindex(name = "year", type = TREE)
privateIntegeryear;
@OverridepublicStringtoString() {
return"Item{" +
"id=" + id +
", name='" + name + '\'' +
", articles=" + articles +
", year=" + year +
'}';
}
publicItem(Integerid, Stringname, List<Integer> articles, Integeryear) {
this.id = id;
this.name = name;
this.articles = articles;
this.year = year;
}
publicstaticvoidmain(String[] args) throwsException {
// Init a database instance and choose the binding (builtin). Configure connection pool size and connection// timeout. Database should be created explicitly via reindexer_tool.// To connect to Reindexer with TLS, use cprotos:// protocol with default port 6535.// Use ReindexerConfiguration#sslSocketFactory to provide a custom SSLSocketFactory.Reindexerdb = ReindexerConfiguration.builder()
.url("cproto://localhost:6534/testdb")
.connectionPoolSize(1)
.requestTimeout(Duration.ofSeconds(30L))
.getReindexer();
// Create new namespace with name 'items', which will store objects of type 'Item'db.openNamespace("items", NamespaceOptions.defaultOptions(), Item.class);
// Generate datasetfor (inti = 0; i < 1000000; i++) {
Randomrandom = newRandom();
db.upsert("items", newItem(
i,
"Vasya",
Arrays.asList(random.nextInt() % 100, random.nextInt() % 100), 2000 + random.nextInt() % 50)
);
}
// Query multiple documents, execute the query and return an iteratorCloseableIterator<Item> iterator = db.query("items", Item.class)
.sort("year", false)
.where("name", EQ, "Vasya")
.where("year", GT, 2020)
.where("articles", SET, 6, 1, 8)
.limit(10)
.offset(0)
.execute();
// Iterate over resultswhile (iterator.hasNext()) {
System.out.println(iterator.next());
}
// Iterator must be closediterator.close();
//Update single itemdb.query("items", Item.class)
.where("id", EQ, 5)
.set("name", "Vova")
.update();
//Update multiple fieldsdb.query("items", Item.class)
.where("id", EQ, 5)
.set("name", "Vova")
.set("year", 2021)
.update();
//Update multiple items itemsdb.query("items", Item.class)
.where("id", LT, 5)
.set("name", "Petya")
.update();
//Drop an item fielddb.query("items", Item.class)
.where("id", EQ, 6)
.drop("name");
}
}

An alternative way to perform queries is to use the "Namespace" object, which can be obtained by opening the namespace using the Reindexer.openNamespace method:

NamespaceitemNamespace = db.openNamespace("items", NamespaceOptions.defaultOptions(), Item.class);
Itemitem = namespace.query()
.where("name", EQ, "Vasya")
.getOne();

Complex Primary Keys and Composite Indexes

A Document can have multiple fields as a primary key. To enable this feature add composite index to object. Composite index is an index that involves multiple fields, it can be used instead of several separate indexes.

// Composite index@Reindex(name = "id+sub_id", isPrimaryKey = true, subIndexes = {"id", "sub_id"})
publicclassItem {
// 'id' is a part of a primary key@Reindex(name = "id")
privateIntegerid;
// 'sub_id' is a part of a primary key@Reindex(name = "sub_id")
privateStringsubId;
}

Query for composite index:

db.query("items", Item.class)
.whereComposite("id+sub_id", EQ, 1, "test")
.execute();

Full text search

Reindexer has internal full text search engine. It can be used for fields with text index. Use the @FullText annotation in code to tune the full text search params of text index. Use it only in conjunction with @Reindex annotation of text type. Full text search query supports either EQ or SET conditions.

publicclassItem {
@Reindex(name = "id")
privateIntegerid;
@Reindex(name = "description", type = TEXT)
@FullText(synonyms = @FullText.Synonym(tokens = {"cpu"}, alternatives = {"processor"}))
privateStringdescription;
}

This query returns all items with words "cpu" or "processor" in description:

db.query("items", Item.class)
.where("description", Query.Condition.EQ, "cpu")
.toList();

This query returns all items with words "cpu" or "processor" but not with word "food" in description:

db.query("items", Item.class)
.where("description", Query.Condition.EQ, "cpu -food")
.toList();

Full text search usage documentation and examples are here.

Joins

Reindexer can join documents from multiple namespaces into a single result:

importru.rt.restream.reindexer.annotations.Transient;
publicclassActor {
@Reindex(name = "id", isPrimaryKey = true)
privateIntegerid;
@Reindex(name = "name")
privateStringname;
@Reindex(name = "is_visible")
privatebooleanvisible;
}
publicclassItemWithJoin {
@Reindex(name = "id", isPrimaryKey = true)
privateIntegerid;
@Reindex(name = "name")
privateStringname;
privateList<Integer> actorsIds;
privateStringactorName;
@TransientprivateList<Actor> joinedActors;
@TransientprivateActorjoinedActor;
}
//Select all items inner join actors on Actor.id in ItemWithJoin.actorIdsCloseableIterator<ItemWithJoin> items = db.query("items_with_join", ItemWithJoin.class)
.join(db.query("actors", Actor.class)
.on("actorsIds", Query.Condition.SET, "id"), "joinedActors")
.execute();
//Select all items inner join visible actors on Actor.id in ItemWithJoin.actorIdsCloseableIterator<ItemWithJoin> items = db.query("items_with_join", ItemWithJoin.class)
.join(db.query("actors", Actor.class).where("is_visible", EQ, true)
.on("actorsIds", Query.Condition.SET, "id"), "joinedActors")
.execute();
//Select all items inner join actors on Actor.name equal ItemWithJoin.actorNameCloseableIterator<ItemWithJoin> items = db.query("items_with_join", ItemWithJoin.class)
.join(db.query("actors", Actor.class)
.on("actorName", Query.Condition.EQ, "name"), "joinedActor")
.execute();

Join query may have from one to several On conditions connected with And (by default), or Or operators:

CloseableIterator<ItemWithJoin> items = db.query("items_with_join",ItemWithJoin.class)
.join(db.query("actors", Actor.class)
.on("actorsIds",Query.Condition.SET,"id")
.on("actorName",Query.Condition.SET,"name"),"joinedActors")
.execute();

An InnerJoin combines data from two namespaces where there is a match on the joining fields in both namespaces. A LeftJoin returns all valid items from the namespaces on the left side of the LeftJoin keyword, along with the values from the table on the right side, or nothing if a matching item doesn't exist.
InnerJoins can be used as a condition in Where clause:

Query<ItemWithJoin> query1 = db.query("items_with_join", ItemWithJoin.class)
.where("id", RANGE, 0, 100)
.or()
.innerJoin(db.query("actors", Actor.class)
.where("name", EQ, "Test")
.on("actorsIds", SET, "id"), "joinedActors")
.or()
.innerJoin(db.query("actors", Actor.class)
.where("id", RANGE, 200, 300)
.on("actorsIds", SET, "id"), "joinedActors")
.execute(); Query<ItemWithJoin> query2 = db.query("items_with_join", ItemWithJoin.class)
.where("id", RANGE, 0, 100)
.or()
.openBracket()
.innerJoin(db.query("actors", Actor.class)
.where("name", EQ, "Test")
.on("actorsIds", SET, "id"), "joinedActors")
.innerJoin(db.query("actors", Actor.class)
.where("id", RANGE, 200, 300)
.on("actorsIds", SET, "id"), "joinedActors")
.closeBracket();
Query<ItemWithJoin> query3 = db.query("items_with_join", ItemWithJoin.class)
.where("id", RANGE, 0, 100)
.or()
.innerJoin(db.query("actors", Actor.class)
.where("id", RANGE, 200, 300)
.on("actorsIds", SET, "id")
.limit(0), "joinedActors");

Note that usually Or operator implements short-circuiting for Where conditions: if the previous condition is true the next one is not evaluated. But in case of InnerJoin it works differently: in query1 (from the example above) both InnerJoin conditions are evaluated despite the result of WhereInt. Limit(0) as part of InnerJoin (query3 from the example above) does not join any data - it works like a filter only to verify conditions.

Query Expressions

Functions

Reindexer provides built-in functions that can be used within WHERE clauses to enable advanced filtering capabilities beyond simple field comparisons.

flat_array_len(field_name)

The flat_array_len function returns the length or cardinality of a specified field, making it particularly useful for filtering based on array sizes or field presence. The flat_array_len function can be used in both SELECT and UPDATE queries.

Behavior by Field Type:

Array Fields: returns the number of elements in the array
Scalar Fields (integers, strings, etc.): always returns 1
Object Fields: always returns 1
Nested Array Elements: returns the count of occurrences when the field is nested within arrays

Examples:

// Find social media posts with between 10 and 50 comments// and at least 3 attached media files.List<Post> posts = db.query("posts", Post.class)
.where(Expression.flatArrayLength("comments"), RANGE, Expression.values(10, 50))
.where(Expression.flatArrayLength("media"), GE, Expression.values(3))
.toList();
// Update field 'size' with flat_array_len function.db.query("posts", Post.class)
.where("id", EQ, 1)
.setExpression("size", Expression.string("flat_array_len(comments)"))
.update();

Notes:

flat_array_len function operates efficiently on indexed fields
Returns 0 if the specified field does not exist in a document
Supports the following comparison operators: (=, >, >=, <, <=, Range, Set)
Can be used in both SELECT and UPDATE queries

now(unit)

The now() function returns the current system timestamp, making it particularly useful for time-based filtering and data synchronization. This function can be used in both SELECT and UPDATE queries.

Arguments:

sec - returns timestamp in seconds (default if no argument is provided)
msec - returns timestamp in milliseconds
usec - returns timestamp in microseconds
nsec - returns timestamp in nanoseconds

// Find events that occurred in the past.List<Event> events = db.query("events", Event.class)
.where(Expression.field("timestamp"), LE, Expression.now(TimeUnit.SECONDS))
.toList();
db.query("items", Item.class)
.where("id", EQ, 42)
.setExpression("updated_at", Expression.string("now(usec)"))
.update();

Notes:

The returned timestamp represents seconds (or subunits) since the Unix epoch (January 1, 1970)
Time resolution depends on the specified unit - use nsec for maximum precision
All instances of now() within a single query share the same value, which is computed at the start of the query execution
Useful for implementing TTL (Time-To-Live) functionality and audit logging

Transactions and batch update

Reindexer supports transactions. Transaction are performs atomic namespace update. There are synchronous and async transaction available. To start transaction method db.beginTransaction() is used. This method creates transaction object, which provides usual Update/Upsert/Insert/Delete interface for application. For RPC clients there is transactions count limitation - each connection can't has more than 1024 opened transactions at the same time.

Synchronous mode

// Create new transaction objectTransaction<Item> tx = db.beginTransaction("items", Item.class);
// Fill transaction objecttx.upsert(newItem(100, "Vasya", Arrays.asList(6, 1, 8), 2019));
tx.upsert(newItem(101, "Vova", Arrays.asList(7, 2, 9), 2020));
tx.query().where("id", EQ, 102).set("name", "Petya").update();
// Apply transactiontx.commit();

Async batch mode

// Create new transaction objectTransaction<Item> tx = db.beginTransaction("items", Item.class);
// Prepare transaction object asynctx.upsertAsync(newItem(100, "Vasya", Arrays.asList(6, 1, 8), 2019));
tx.upsertAsync(newItem(101, "Vova", Arrays.asList(7, 2, 9), 2020))
.thenAccept(item -> processItem(item))
.exceptionally(e -> handleError(e));
// Wait for async operations done, and apply transactiontx.commit();

The return value of tx.upsertAsync is CompletableFuture, which will be completed after receiving server response. Also, if any error occurred during prepare process, then tx.commit should return an error. So it is enough, to check error returned by tx.commit - to be sure, that all data has been successfully committed or not.

Transactions commit strategies

Depends on amount changes in transaction there are 2 possible Commit strategies:

Locked atomic update. Reindexer locks namespace and applying all changes under common lock. This mode is used with small amounts of changes.
Copy & atomic replace. In this mode Reindexer makes namespace's snapshot, applying all changes to this snapshot, and atomically replaces namespace without lock.

Implementation notes

Transaction object is not thread safe and can't be used from different threads.
Transaction object holds Reindexer's resources, therefore application should explicitly call tx.rollback or tx.commit, otherwise resources will leak.
It is safe to call tx.rollback after tx.commit.
It is possible to call Query from transaction by call tx.query().execute(); .... Only read-committed isolation is available. Changes made in active transaction is invisible to current and another transactions.

Observability support

For metrics and traces, reindexer-java uses Micrometer Observation. To enable observation, you need to provide an ObservationRegistry to the ReindexerConfiguration.

The following example shows how to configure observation for reindexer-java using Prometheus:

Add micrometer-registry-prometheus dependency to the pom.xml:

<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
<version>${micrometer.version}</version>
</dependency>

Provide an ObservationRegistry implementation to the ReindexerConfiguration:

// 1. Initialize Prometheus Meter Registry:PrometheusMeterRegistryprometheusRegistry = newPrometheusMeterRegistry(PrometheusConfig.DEFAULT);
// 2. Initialize Observation Registry:ObservationRegistryobservationRegistry = ObservationRegistry.create();
// 3. Bridge them together using DefaultMeterObservationHandler:observationRegistry.observationConfig()
.observationHandler(newDefaultMeterObservationHandler(prometheusRegistry));
// 4. Provide an ObservationRegistry to ReindexerConfiguration:Reindexerdb = ReindexerConfiguration.builder()
.url("cproto://localhost:6534/testdb")
.connectionPoolSize(8)
.requestTimeout(Duration.ofSeconds(30L))
.observationRegistry(observationRegistry)
.getReindexer();

Collected metrics and traces

All Reindexer RPC commands executed by reindexer-java are instrumented with Micrometer.

The following low cardinality key values are added to observations:

db.system.name - the name of the database system, always reindexer
db.command.name - the name of the RPC command being executed, e.g., selectQuery
db.namespace - the database name e.g., test_db
db.collection.name - the collection name that the RPC command is executed on e.g., items
network.transport - the protocol used for the RPC command e.g., cproto, cprotos
server.address - the host of the Reindexer node that the RPC command is sent to e.g., localhost
server.port - the port of the Reindexer node that the RPC command is sent to e.g., 6534
code.execution_type - the code execution type e.g., SYNC, ASYNC
db.response.status_code - the Reindexer response status code

Additionally, the following high-cardinality key values are added to traces:

thread.id - ID of the thread executing the RPC command
thread.name - name of the thread executing the RPC command
db.reindexer.tx_id - ID of the Reindexer transaction associated with the RPC command, when applicable
db.reindexer.rq_id - ID of the Reindexer request associated with the RPC command

Development notes

To run tests locally, you need to install Reindexer using a package manager for your OS. Cprotos protocol tests require Reindexer to be built with TLS support, and ssl certificate and key must be placed in src/test/resources, to generate a new valid certificate run the following commands:

cd src/test/resources
# Generates a certificate key (not needed if you already have one).
openssl genrsa -out builtin-server.key 2048
# Generates a self-signed certificate using the provided key;# Prompts to fill certificate information e.g. CN=localhost;# The certificate will be valid for 10 years.
openssl req -new -x509 -key builtin-server.key -out builtin-server.crt -days 3650
# Import the certificate into the Java keystore and save it as a JKS file.
keytool -importcert -alias builtin-server -file builtin-server.crt -keystore builtin-server.jks -storepass password -noprompt

Name		Name	Last commit message	Last commit date
Latest commit History 378 Commits
.github/workflows		.github/workflows
builtin-adapter		builtin-adapter
src		src
.gitignore		.gitignore
LICENSE		LICENSE
README.md		README.md
header.txt		header.txt
pom.xml		pom.xml

Navigation Menu

Uh oh!

Repository files navigation

rx-connector

Maven

Usage

Complex Primary Keys and Composite Indexes

Full text search

Joins

Query Expressions

Functions

flat_array_len(field_name)

now(unit)

Transactions and batch update

Synchronous mode

Async batch mode

Transactions commit strategies

Implementation notes

Observability support

Collected metrics and traces

Development notes

About

Releases

Packages

Used by

Contributors

Languages

Uh oh!

Folders and files

Latest commit

History

Repository files navigation

rx-connector

Maven

Usage

Complex Primary Keys and Composite Indexes

Full text search

Joins

Query Expressions

Functions

flat_array_len(field_name)

now(unit)

Transactions and batch update

Synchronous mode

Async batch mode

Transactions commit strategies

Implementation notes

Observability support

Collected metrics and traces

Development notes

About

Resources

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages