Nested joins

builtin-adapter/BuiltinAdapter.cpp

@@ -91,20 +91,27 @@ JNIEXPORT jlong JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAda
     return init_reindexer();
 }
 
+JNIEXPORT jstring JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_version(JNIEnv *env, jobject) {
+    return env->NewStringUTF(reindexer_version());
+}
+
 JNIEXPORT void JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_destroy(JNIEnv *, jobject,
                                                                                             jlong rx) {
     destroy_reindexer(rx);
 }
 
 JNIEXPORT jobject JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_connect(JNIEnv *env, jobject,
                                                                                                jlong rx, jstring path,
-                                                                                               jstring version) {
+                                                                                               jstring version,
+                                                                                               jint queryFormatVersion) {
     reindexer_string dsn = rx_string(env, path);
     reindexer_string vers = rx_string(env, version);
+    int64_t capabilities = kBindingCapabilityResultsWithShardIDs | kBindingCapabilityComplexRank;
+    if (queryFormatVersion == QueryFormatV2) {
+        capabilities |= kBindingCapabilityQueryFormatV2;
+    }
     reindexer_error error = reindexer_connect(rx, dsn, ConnectOpts(), vers, BindingCapabilities(
-        kBindingCapabilityResultsWithShardIDs
-        | kBindingCapabilityComplexRank
-        | kBindingCapabilityQueryFormatV2));
+        capabilities));
     env->ReleaseStringUTFChars(path, reinterpret_cast<const char *>(dsn.p));
     env->ReleaseStringUTFChars(version, reinterpret_cast<const char *>(vers.p));
     return j_res(env, error);

builtin-adapter/BuiltinAdapter.h

@@ -24,10 +24,13 @@ extern "C" {
 
 JNIEXPORT jlong JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_init(JNIEnv *, jobject);
 
+JNIEXPORT jstring JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_version(JNIEnv *, jobject);
+
 JNIEXPORT void JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_destroy(JNIEnv *, jobject, jlong);
 
 JNIEXPORT jobject JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_connect(JNIEnv *, jobject, jlong,
-                                                                                               jstring, jstring);
+                                                                                               jstring, jstring,
+                                                                                               jint);
 
 JNIEXPORT jobject JNICALL Java_ru_rt_restream_reindexer_binding_builtin_BuiltinAdapter_openNamespace(JNIEnv *, jobject,
                                                                                                      jlong, jlong,

src/main/java/ru/rt/restream/reindexer/Query.java

@@ -1317,27 +1317,23 @@ public byte[] bytes() {
     }
 
     private byte[] toSubQueryBytes(int formatVersion) {
-        byte[] queryBytes = getQueryBytes(formatVersion);
-        if (formatVersion == QUERY_FORMAT_V2 || hasNestedJoins()) {
-            ByteBuffer copy = new ByteBuffer(queryBytes);
-            copy.putVarUInt32(QUERY_END);
-            copy.putVarUInt32(0);
-            copy.putVarUInt32(0);
-            return copy.bytes();
+        if (formatVersion == QUERY_FORMAT_V2) {
+            return serializeQuery(formatVersion);
+        }
+        if (!joinQueries.isEmpty() || !mergeQueries.isEmpty()) {
+            throw new IllegalStateException("Join and merge queries in subquery are not supported by QueryFormatV1");
         }
-        return queryBytes;
+        return getQueryBytes(formatVersion);
     }
 
     private byte[] toExecutableBytes() {
         int formatVersion = reindexer.getBinding().queryFormatVersion();
-        ByteBuffer queryBuffer = new ByteBuffer(getQueryBytes(formatVersion));
-        queryBuffer.putVarUInt32(QUERY_END);
         if (formatVersion == QUERY_FORMAT_V2) {
-            appendJoinQueries(queryBuffer, new ArrayList<>(), formatVersion);
-            appendMergeQueries(queryBuffer, new ArrayList<>(), formatVersion);
-        } else {
-            appendJoinQueriesV1(queryBuffer, false);
+            return serializeQuery(formatVersion);
         }
+        ByteBuffer queryBuffer = new ByteBuffer(getQueryBytes(formatVersion));
+        queryBuffer.putVarUInt32(QUERY_END);
+        appendJoinQueriesV1(queryBuffer, false);
         return queryBuffer.bytes();
     }
 
@@ -1367,6 +1363,14 @@ private void appendMergeQueries(ByteBuffer target, List<ReindexerNamespace<?>> t
         }
     }
 
+    private byte[] serializeQuery(int formatVersion) {
+        ByteBuffer queryBuffer = new ByteBuffer(getQueryBytes(formatVersion));
+        queryBuffer.putVarUInt32(QUERY_END);
+        appendJoinQueries(queryBuffer, new ArrayList<>(), formatVersion);
+        appendMergeQueries(queryBuffer, new ArrayList<>(), formatVersion);
+        return queryBuffer.bytes();
+    }
+
     private void appendQuery(ByteBuffer target, Query<?> query, int queryJoinType,
                              List<ReindexerNamespace<?>> targetNamespaces, int formatVersion) {
         if (queryJoinType != MERGE) {
@@ -1382,7 +1386,7 @@ private void appendQuery(ByteBuffer target, Query<?> query, int queryJoinType,
     }
 
     private void appendJoinQueriesV1(ByteBuffer target, boolean appendNamespaces) {
-        if (hasNestedJoins()) {
+        if (hasNestedQueries()) {
             throw new IllegalStateException("Nested joins are not supported by QueryFormatV1");
         }
         for (Query<?> joinQuery : joinQueries) {
@@ -1410,14 +1414,14 @@ private void appendMergeQueriesV1(ByteBuffer target) {
         }
     }
 
-    private boolean hasNestedJoins() {
+    private boolean hasNestedQueries() {
         for (Query<?> joinQuery : joinQueries) {
-            if (!joinQuery.joinQueries.isEmpty() || joinQuery.hasNestedJoins()) {
+            if (!joinQuery.joinQueries.isEmpty() || !joinQuery.mergeQueries.isEmpty() || joinQuery.hasNestedQueries()) {
                 return true;
             }
         }
         for (Query<?> mergeQuery : mergeQueries) {
-            if (mergeQuery.hasNestedJoins()) {
+            if (!mergeQuery.mergeQueries.isEmpty() || mergeQuery.hasNestedQueries()) {
                 return true;
             }
         }

src/main/java/ru/rt/restream/reindexer/QueryResultIterator.java

@@ -193,10 +193,19 @@ private void readJoinedItems(Object item, Query<?> queryContext, int nsId) {
             for (int i = 0; i < itemsCount; i++) {
                 subItems.add(readItem(joinedNamespace, joinedItemReader, joinQuery));
             }
-            subItemsMap.computeIfAbsent(queryContext.getJoinFields().get(joinedField), field -> new ArrayList<>())
-                    .addAll(subItems);
+
+            String joinField = queryContext.getJoinFields().get(joinedField);
+            List<Object> fieldSubItems = subItemsMap.get(joinField);
+            if (fieldSubItems == null) {
+                fieldSubItems = new ArrayList<>();
+                subItemsMap.put(joinField, fieldSubItems);
+            }
+            fieldSubItems.addAll(subItems);
+        }
+
+        for (Map.Entry<String, List<Object>> entry : subItemsMap.entrySet()) {
+            writeJoinResult(item, entry.getKey(), entry.getValue());
         }
-        subItemsMap.forEach((key, value) -> writeJoinResult(item, key, value));
     }
 
     private void readJoinedItemsV1(Object item, int nsId) {

src/main/java/ru/rt/restream/reindexer/binding/Binding.java

@@ -214,6 +214,15 @@ default int queryFormatVersion() {
         return Consts.QUERY_FORMAT_V1;
     }
 
+    /**
+     * Returns true if connected Reindexer server supports nested join queries.
+     *
+     * @return true if connected Reindexer server supports nested join queries
+     */
+    default boolean supportsNestedJoinQueries() {
+        return false;
+    }
+
     /**
      * Closes binding to Reindexer instance.
      */

src/main/java/ru/rt/restream/reindexer/binding/QueryResultReader.java

@@ -75,8 +75,7 @@ public QueryResult read(byte[] rawQueryResult, int queryFormatVersion) {
         if (queryFormatVersion == QUERY_FORMAT_V2) {
             long format = buffer.getVarUInt();
             if (format != QUERY_FORMAT_V2) {
-                String errorMessage = String.format("QueryResults format version='%d' is not supported", format);
-                throw new RuntimeException(errorMessage);
+                buffer.rewind();
             }
         }
         QueryResult queryResult = getQueryResultWithFlags(buffer.getVarUInt());

src/main/java/ru/rt/restream/reindexer/binding/builtin/Builtin.java

@@ -45,6 +45,12 @@ public class Builtin implements Binding {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(Builtin.class);
 
+    private static final int NESTED_JOIN_QUERIES_MIN_MAJOR = 5;
+
+    private static final int NESTED_JOIN_QUERIES_MIN_MINOR = 16;
+
+    private static final int NESTED_JOIN_QUERIES_MIN_PATCH = 0;
+
     private final AtomicLong next = new AtomicLong(0L);
 
     private final Gson gson = new GsonBuilder()
@@ -57,6 +63,10 @@ public class Builtin implements Binding {
 
     private final Duration timeout;
 
+    private final boolean supportsNestedJoinQueries;
+
+    private final int queryFormatVersion;
+
     /**
      * Creates an instance.
      *
@@ -67,9 +77,11 @@ public Builtin(URI uri, Duration requestTimeout) {
         adapter = new BuiltinAdapter();
         timeout = requestTimeout;
         rx = adapter.init();
+        queryFormatVersion = QUERY_FORMAT_V2;
+        supportsNestedJoinQueries = isNestedJoinQueriesSupported(adapter.version());
         String path = uri.getPath();
         try {
-            ReindexerResponse response = adapter.connect(rx, path, REINDEXER_VERSION);
+            ReindexerResponse response = adapter.connect(rx, path, REINDEXER_VERSION, queryFormatVersion);
             checkResponse(response);
         } catch (Exception e) {
             LOGGER.error("rx: connect error", e);
@@ -89,6 +101,8 @@ public Builtin(BuiltinAdapter adapter, long rx, Duration timeout) {
         this.adapter = adapter;
         this.rx = rx;
         this.timeout = timeout;
+        queryFormatVersion = QUERY_FORMAT_V2;
+        supportsNestedJoinQueries = isNestedJoinQueriesSupported(adapter.version());
     }
 
     @Override
@@ -152,15 +166,15 @@ public RequestContext select(String query, boolean asJson, int fetchCount, long[
         ReindexerResponse response = adapter.select(rx, next.getAndIncrement(), timeout.toMillis(), query, asJson,
                 ptVersions);
         checkResponse(response);
-        return new BuiltinRequestContext(response);
+        return new BuiltinRequestContext(response, queryFormatVersion);
     }
 
     @Override
     public RequestContext selectQuery(byte[] queryData, int fetchCount, long[] ptVersions, boolean asJson) {
         ReindexerResponse response = adapter.selectQuery(rx, next.getAndIncrement(), timeout.toMillis(), queryData,
                 ptVersions, asJson);
         checkResponse(response);
-        return new BuiltinRequestContext(response);
+        return new BuiltinRequestContext(response, queryFormatVersion);
     }
 
     @Override
@@ -187,7 +201,7 @@ public TransactionContext beginTx(String namespaceName) {
                 txId = (long) arg;
             }
         }
-        return new BuiltinTransactionContext(adapter, rx, txId, next::getAndIncrement, timeout);
+        return new BuiltinTransactionContext(adapter, rx, txId, next::getAndIncrement, timeout, queryFormatVersion);
     }
 
     @Override
@@ -210,7 +224,35 @@ private void checkResponse(ReindexerResponse response) {
 
     @Override
     public int queryFormatVersion() {
-        return QUERY_FORMAT_V2;
+        return queryFormatVersion;
+    }
+
+    @Override
+    public boolean supportsNestedJoinQueries() {
+        return supportsNestedJoinQueries;
+    }
+
+    private boolean isNestedJoinQueriesSupported(String version) {
+        int[] parsedVersion = parseVersion(version);
+        if (parsedVersion[0] != NESTED_JOIN_QUERIES_MIN_MAJOR) {
+            return parsedVersion[0] > NESTED_JOIN_QUERIES_MIN_MAJOR;
+        }
+        if (parsedVersion[1] != NESTED_JOIN_QUERIES_MIN_MINOR) {
+            return parsedVersion[1] > NESTED_JOIN_QUERIES_MIN_MINOR;
+        }
+        return parsedVersion[2] >= NESTED_JOIN_QUERIES_MIN_PATCH;
+    }
+
+    private int[] parseVersion(String version) {
+        String normalized = version.startsWith("v") ? version.substring(1) : version;
+        String[] parts = normalized.split("\\D+");
+        int[] result = new int[3];
+        for (int i = 0; i < result.length && i < parts.length; i++) {
+            if (!parts[i].isEmpty()) {
+                result[i] = Integer.parseInt(parts[i]);
+            }
+        }
+        return result;
     }
 
     @Override

src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinAdapter.java

@@ -72,6 +72,13 @@ private static void loadLibrary(String fileName) throws IOException {
      */
     public native long init();
 
+    /**
+     * Returns native Reindexer library version.
+     *
+     * @return the native Reindexer library version
+     */
+    public native String version();
+
     /**
      * Destroys Reindexer instance.
      *
@@ -135,9 +142,10 @@ private static void loadLibrary(String fileName) throws IOException {
      * @param rx      the Reindexer instance pointer
      * @param path    the Reindexer's database path
      * @param version the Reindexer's version
+     * @param queryFormatVersion the query serialization format version
      * @return the {@link ReindexerResponse} to use
      */
-    public native ReindexerResponse connect(long rx, String path, String version);
+    public native ReindexerResponse connect(long rx, String path, String version, int queryFormatVersion);
 
     /**
      * Opens a namespace.

src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinRequestContext.java

@@ -22,8 +22,6 @@
 import ru.rt.restream.reindexer.binding.RequestContext;
 import ru.rt.restream.reindexer.util.NativeUtils;
 
-import static ru.rt.restream.reindexer.binding.Consts.QUERY_FORMAT_V2;
-
 /**
  * A request context which is holds a {@link QueryResult},
  * the {@link #fetchResults(int, int)} method is NOOP since Builtin does not support it.
@@ -37,7 +35,7 @@ public class BuiltinRequestContext implements RequestContext {
      *
      * @param response the {@link ReindexerResponse} to use
      */
-    public BuiltinRequestContext(ReindexerResponse response) {
+    public BuiltinRequestContext(ReindexerResponse response, int queryFormatVersion) {
         long resultsPtr = 0L;
         byte[] rawQueryResult = new byte[0];
         Object[] arguments = response.getArguments();
@@ -54,7 +52,7 @@ public BuiltinRequestContext(ReindexerResponse response) {
             }
         }
         QueryResultReader reader = new QueryResultReader();
-        queryResult = reader.read(rawQueryResult, QUERY_FORMAT_V2);
+        queryResult = reader.read(rawQueryResult, queryFormatVersion);
         queryResult.setResultsPtr(resultsPtr);
     }
 

src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinTransactionContext.java

@@ -46,6 +46,8 @@ public class BuiltinTransactionContext implements TransactionContext {
 
     private final Duration timeout;
 
+    private final int queryFormatVersion;
+
     /**
      * Creates an instance.
      *
@@ -56,12 +58,13 @@ public class BuiltinTransactionContext implements TransactionContext {
      * @param timeout       the execution timeout
      */
     public BuiltinTransactionContext(BuiltinAdapter adapter, long rx, long transactionId,
-                                     Supplier<Long> next, Duration timeout) {
+                                     Supplier<Long> next, Duration timeout, int queryFormatVersion) {
         this.adapter = adapter;
         this.rx = rx;
         this.transactionId = transactionId;
         this.next = next;
         this.timeout = timeout;
+        this.queryFormatVersion = queryFormatVersion;
     }
 
     @Override
@@ -91,7 +94,7 @@ private ReindexerResponse modifyItemInternal(byte[] data, int format, int mode,
     public RequestContext selectQuery(byte[] queryData, int fetchCount, long[] ptVersions, boolean asJson) {
         ReindexerResponse response = adapter.selectQuery(rx, next.get(), timeout.toMillis(), queryData, ptVersions, asJson);
         checkResponse(response);
-        return new BuiltinRequestContext(response);
+        return new BuiltinRequestContext(response, queryFormatVersion);
     }
 
     @Override

src/main/java/ru/rt/restream/reindexer/binding/builtin/server/BuiltinServer.java

@@ -196,6 +196,11 @@ public int queryFormatVersion() {
         return builtin.queryFormatVersion();
     }
 
+    @Override
+    public boolean supportsNestedJoinQueries() {
+        return builtin.supportsNestedJoinQueries();
+    }
+
     @Override
     public void close() {
         ReindexerResponse response = adapter.stopServer(svc);

src/main/java/ru/rt/restream/reindexer/binding/cproto/Connection.java

@@ -59,6 +59,15 @@ default int queryFormatVersion() {
         return ru.rt.restream.reindexer.binding.Consts.QUERY_FORMAT_V1;
     }
 
+    /**
+     * Returns true if connected Reindexer server supports nested join queries.
+     *
+     * @return true if connected Reindexer server supports nested join queries
+     */
+    default boolean supportsNestedJoinQueries() {
+        return false;
+    }
+
     /**
      * Closes the connection.
      */