Merge pull request #134 from Restream/feature/v5.13.0-support

Add support for v5.13.0 reindexer API

Author: MadSchemas

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;
 }

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,

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});
         }
     }
 

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.

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);
     }
 

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.

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

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