[fea] Add support for v5.13.0 reindexer API

MadSchemas · MadSchemas · commit e6d4976840b3 · 2026-04-29T16:23:45.000+03:00
diff --git a/builtin-adapter/BuiltinAdapter.cpp b/builtin-adapter/BuiltinAdapter.cpp
@@ -282,9 +282,13 @@ JNIEXPORT jobject JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinA
                                                                                                    jlong rx,
                                                                                                    jlong ctxId,
                                                                                                    jlong timeout,
-                                                                                                   jbyteArray data) {
+                                                                                                   jbyteArray data,
+                                                                                                   jlongArray versions) {
+    auto ptVersions = reinterpret_cast<int32_t *>(env->GetLongArrayElements(versions, nullptr));
+    int ptVersionsCount = env->GetArrayLength(versions);
     reindexer_buffer bufferData = rx_buffer(env, data);
-    jobject res = j_res(env, reindexer_update_query(rx, bufferData, rx_ctx(ctxId, timeout)));
+    jobject res = j_res(env, reindexer_update_query(rx, bufferData, ptVersions, ptVersionsCount, rx_ctx(ctxId, timeout)));
+    env->ReleaseLongArrayElements(versions, reinterpret_cast<jlong *>(ptVersions), 0);
     env->ReleaseByteArrayElements(data, reinterpret_cast<jbyte *>(bufferData.data), 0);
     return res;
 }
diff --git a/builtin-adapter/BuiltinAdapter.h b/builtin-adapter/BuiltinAdapter.h
@@ -91,7 +91,8 @@ JNIEXPORT jobject JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinA
 
 JNIEXPORT jobject JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_updateQuery(JNIEnv *, jobject,
                                                                                                    jlong, jlong, jlong,
-                                                                                                   jbyteArray);
+                                                                                                   jbyteArray,
+                                                                                                   jlongArray);
 
 JNIEXPORT jobject JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_updateQueryTx(JNIEnv *, jobject,
                                                                                                      jlong, jlong,
diff --git a/src/main/java/ru/rt/restream/reindexer/Query.java b/src/main/java/ru/rt/restream/reindexer/Query.java
@@ -1074,8 +1074,7 @@ private long[] prepareQueryAndGetPayloadTypesVersions() {
 
         return namespaces.stream()
                 .map(ReindexerNamespace::getPayloadType)
-                .map(pt -> pt == null ? 0 : pt.getStateToken())
-                .mapToLong(Integer::longValue)
+                .mapToLong(pt -> pt == null ? 0 : (pt.getVersion() ^ pt.getStateToken()))
                 .toArray();
     }
 
@@ -1245,7 +1244,11 @@ public void update() {
         if (transactionContext != null) {
             transactionContext.updateQuery(buffer.bytes());
         } else {
-            reindexer.getBinding().updateQuery(buffer.bytes());
+            // There are no support for inner joins for update-queries in Java binding,
+            // so we are using single pt version
+            PayloadType pt = namespace.getPayloadType();
+            long tmVersion = pt == null ? 0 : (pt.getVersion() ^ pt.getStateToken());
+            reindexer.getBinding().updateQuery(buffer.bytes(), new long[]{tmVersion});
         }
     }
 
diff --git a/src/main/java/ru/rt/restream/reindexer/binding/Binding.java b/src/main/java/ru/rt/restream/reindexer/binding/Binding.java
@@ -173,8 +173,9 @@ public interface Binding {
      * Invoke update query.
      *
      * @param queryData encoded query data (selected indexes, predicates, etc)
+     * @param ptVersions payload type state tokens
      */
-    void updateQuery(byte[] queryData);
+    void updateQuery(byte[] queryData, long[] ptVersions);
 
     /**
      * Starts a transaction for the given namespace name.
diff --git a/src/main/java/ru/rt/restream/reindexer/binding/builtin/Builtin.java b/src/main/java/ru/rt/restream/reindexer/binding/builtin/Builtin.java
@@ -169,8 +169,8 @@ public void deleteQuery(byte[] queryData) {
     }
 
     @Override
-    public void updateQuery(byte[] queryData) {
-        ReindexerResponse response = adapter.updateQuery(rx, next.getAndIncrement(), timeout.toMillis(), queryData);
+    public void updateQuery(byte[] queryData, long[] ptVersions) {
+        ReindexerResponse response = adapter.updateQuery(rx, next.getAndIncrement(), timeout.toMillis(), queryData, ptVersions);
         checkResponse(response);
     }
 
diff --git a/src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinAdapter.java b/src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinAdapter.java
@@ -285,7 +285,7 @@ public native ReindexerResponse openNamespace(long rx, long ctxId, long timeout,
      * @param timeout  the execution timeout
      * @param query    the SQL-query string
      * @param asJson   'true' if response should be serialized in JSON format, defaults to CJSON
-     * @param versions the versions
+     * @param versions the tagsmatcher versions
      * @return the {@link ReindexerResponse} to use
      */
     public native ReindexerResponse select(long rx, long ctxId, long timeout, String query, boolean asJson,
@@ -319,9 +319,10 @@ public native ReindexerResponse select(long rx, long ctxId, long timeout, String
      * @param ctxId   the context id
      * @param timeout the execution timeout
      * @param data    the query payload (i.e. predicates, joins etc)
+     * @param versions the tagsmatcher versions
      * @return the {@link ReindexerResponse} to use
      */
-    public native ReindexerResponse updateQuery(long rx, long ctxId, long timeout, byte[] data);
+    public native ReindexerResponse updateQuery(long rx, long ctxId, long timeout, byte[] data, long[] versions);
 
     /**
      * Executes update query in the transaction.
diff --git a/src/main/java/ru/rt/restream/reindexer/binding/builtin/server/BuiltinServer.java b/src/main/java/ru/rt/restream/reindexer/binding/builtin/server/BuiltinServer.java
@@ -172,8 +172,8 @@ public void deleteQuery(byte[] queryData) {
     }
 
     @Override
-    public void updateQuery(byte[] queryData) {
-        builtin.updateQuery(queryData);
+    public void updateQuery(byte[] queryData, long[] ptVersions) {
+        builtin.updateQuery(queryData, ptVersions);
     }
 
     @Override
diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/Cproto.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/Cproto.java
@@ -165,8 +165,9 @@ public void deleteQuery(byte[] queryData) {
     }
 
     @Override
-    public void updateQuery(byte[] queryData) {
-        rpcCallNoResults(UPDATE_QUERY, queryData);
+    public void updateQuery(byte[] queryData, long[] ptVersions) {
+        int flags = Consts.RESULTS_PURE | Consts.RESULTS_WITH_PAYLOAD_TYPES;
+        rpcCallNoResults(UPDATE_QUERY, queryData, flags, ptVersions);
     }
 
     @Override

Original file line number	Diff line number	Diff line change
`@@ -1074,8 +1074,7 @@ private long[] prepareQueryAndGetPayloadTypesVersions() {`
`1074`	`1074`
`1075`	`1075`	`return namespaces.stream()`
`1076`	`1076`	`.map(ReindexerNamespace::getPayloadType)`
`1077`		`- .map(pt -> pt == null ? 0 : pt.getStateToken())`
`1078`		`- .mapToLong(Integer::longValue)`
	`1077`	`+ .mapToLong(pt -> pt == null ? 0 : (pt.getVersion() ^ pt.getStateToken()))`
`1079`	`1078`	`.toArray();`
`1080`	`1079`	`}`
`1081`	`1080`
`@@ -1245,7 +1244,11 @@ public void update() {`
`1245`	`1244`	`if (transactionContext != null) {`
`1246`	`1245`	`transactionContext.updateQuery(buffer.bytes());`
`1247`	`1246`	`} else {`
`1248`		`- reindexer.getBinding().updateQuery(buffer.bytes());`
	`1247`	`+ // There are no support for inner joins for update-queries in Java binding,`
	`1248`	`+ // so we are using single pt version`
	`1249`	`+ PayloadType pt = namespace.getPayloadType();`
	`1250`	`+ long tmVersion = pt == null ? 0 : (pt.getVersion() ^ pt.getStateToken());`
	`1251`	`+ reindexer.getBinding().updateQuery(buffer.bytes(), new long[]{tmVersion});`
`1249`	`1252`	`}`
`1250`	`1253`	`}`
`1251`	`1254`
Original file line number	Diff line number	Diff line change
`@@ -169,8 +169,8 @@ public void deleteQuery(byte[] queryData) {`
`169`	`169`	`}`
`170`	`170`
`171`	`171`	`@Override`
`172`		`- public void updateQuery(byte[] queryData) {`
`173`		`- ReindexerResponse response = adapter.updateQuery(rx, next.getAndIncrement(), timeout.toMillis(), queryData);`
	`172`	`+ public void updateQuery(byte[] queryData, long[] ptVersions) {`
	`173`	`+ ReindexerResponse response = adapter.updateQuery(rx, next.getAndIncrement(), timeout.toMillis(), queryData, ptVersions);`
`174`	`174`	`checkResponse(response);`
`175`	`175`	`}`
`176`	`176`
Original file line number	Diff line number	Diff line change
`@@ -172,8 +172,8 @@ public void deleteQuery(byte[] queryData) {`
`172`	`172`	`}`
`173`	`173`
`174`	`174`	`@Override`
`175`		`- public void updateQuery(byte[] queryData) {`
`176`		`- builtin.updateQuery(queryData);`
	`175`	`+ public void updateQuery(byte[] queryData, long[] ptVersions) {`
	`176`	`+ builtin.updateQuery(queryData, ptVersions);`
`177`	`177`	`}`
`178`	`178`
`179`	`179`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -165,8 +165,9 @@ public void deleteQuery(byte[] queryData) {`
`165`	`165`	`}`
`166`	`166`
`167`	`167`	`@Override`
`168`		`- public void updateQuery(byte[] queryData) {`
`169`		`- rpcCallNoResults(UPDATE_QUERY, queryData);`
	`168`	`+ public void updateQuery(byte[] queryData, long[] ptVersions) {`
	`169`	`+ int flags = Consts.RESULTS_PURE \| Consts.RESULTS_WITH_PAYLOAD_TYPES;`
	`170`	`+ rpcCallNoResults(UPDATE_QUERY, queryData, flags, ptVersions);`
`170`	`171`	`}`
`171`	`172`
`172`	`173`	`@Override`