Finalize nested joins support

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

@@ -213,14 +213,11 @@ public enum Condition {
 
     private Query<?> root;
 
-    private final int queryFormatVersion;
-
     Query(Reindexer reindexer, ReindexerNamespace<T> namespace, TransactionContext transactionContext) {
         logBuilder.namespace(namespace.getName());
         this.reindexer = reindexer;
         this.namespace = namespace;
         this.transactionContext = transactionContext;
-        this.queryFormatVersion = reindexer.getBinding().queryFormatVersion();
         buffer.putUInt8(0);
         buffer.putVString(namespace.getName());
     }
@@ -249,7 +246,9 @@ public Query<T> selectAllFields() {
     }
 
     /**
-     * Inner joins 2 queries, alias for innerJoin.
+     * Inner joins 2 queries, alias for {@link #innerJoin(Query, String)}.
+     * <p>
+     * Nested joins are supported the same way as in {@link #innerJoin(Query, String)}.
      *
      * @param <J>       type of joined items
      * @param joinQuery query to join
@@ -264,6 +263,20 @@ public <J> Query<T> join(Query<J> joinQuery, String field) {
 
     /**
      * Inner joins 2 queries.
+     * <p>
+     * {@code joinQuery} may itself contain nested inner/left joins; attach those joins to the subquery
+     * before passing it here. Nested joins require QueryFormatV2 (always used by builtin; negotiated for
+     * cproto). With QueryFormatV1 {@link #execute()} throws {@link IllegalStateException}.
+     * <p>
+     * Call {@link #on(String, Condition, String)} on the join subquery, not on this query:
+     * <pre>{@code
+     * Query<Location> locations = db.query("locations", Location.class)
+     *         .on("locationId", EQ, "id");
+     * Query<Author> authors = db.query("authors", Author.class)
+     *         .innerJoin(locations, "locations")
+     *         .on("authorId", EQ, "id");
+     * db.query("books", Book.class).innerJoin(authors, "authors").toList();
+     * }</pre>
      *
      * @param <J>       type of joined items
      * @param joinQuery query to join
@@ -283,6 +296,8 @@ public <J> Query<T> innerJoin(Query<J> joinQuery, String field) {
 
     /**
      * Left joins 2 queries.
+     * <p>
+     * Nested joins are supported the same way as in {@link #innerJoin(Query, String)}.
      *
      * @param <J>       type of joined items
      * @param joinQuery query to join
@@ -316,10 +331,13 @@ private <J> Query<T> join(Query<J> joinQuery, String field, int joinType) {
 
     /**
      * Specify the join condition.
+     * <p>
+     * Call this on the join subquery (the right side), before or after attaching it with
+     * {@link #innerJoin(Query, String)} / {@link #leftJoin(Query, String)}.
      *
-     * @param joinField the join field of the right side of the join
+     * @param joinField the join field of the left side of the join
      * @param condition the joining condition. {@link Condition}
-     * @param joinIndex the join index of the left side of join
+     * @param joinIndex the join index of the right side of the join
      * @return the {@link Query} for further customizations
      */
     public Query<T> on(String joinField, Condition condition, String joinIndex) {
@@ -1050,6 +1068,7 @@ private byte[] buildSelectQueryBytes() {
         }
 
         int formatVersion = reindexer.getBinding().queryFormatVersion();
+        ensureNoMergeNestedInJoin();
         ByteBuffer queryBuffer = new ByteBuffer(getQueryBytes(formatVersion));
         queryBuffer.putVarUInt32(QUERY_END);
         if (formatVersion == QUERY_FORMAT_V2) {
@@ -1313,23 +1332,29 @@ String getSql() {
     }
 
     public byte[] bytes() {
-        return toSubQueryBytes(queryFormatVersion);
+        return toSubQueryBytes(reindexer.getBinding().queryFormatVersion());
     }
 
     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 (!joinQueries.isEmpty()) {
+            throw new IllegalStateException("Join cannot be in subquery");
         }
-        return queryBytes;
+        if (!mergeQueries.isEmpty()) {
+            throw new IllegalStateException("Merge cannot be in subquery");
+        }
+        if (formatVersion == QUERY_FORMAT_V2) {
+            ByteBuffer queryBuffer = new ByteBuffer(getQueryBytes(formatVersion));
+            queryBuffer.putVarUInt32(QUERY_END);
+            queryBuffer.putVarUInt32(0);
+            queryBuffer.putVarUInt32(0);
+            return queryBuffer.bytes();
+        }
+        return getQueryBytes(formatVersion);
     }
 
     private byte[] toExecutableBytes() {
         int formatVersion = reindexer.getBinding().queryFormatVersion();
+        ensureNoMergeNestedInJoin();
         ByteBuffer queryBuffer = new ByteBuffer(getQueryBytes(formatVersion));
         queryBuffer.putVarUInt32(QUERY_END);
         if (formatVersion == QUERY_FORMAT_V2) {
@@ -1424,6 +1449,26 @@ private boolean hasNestedJoins() {
         return false;
     }
 
+    private void ensureNoMergeNestedInJoin() {
+        if (hasMergeNestedInJoin()) {
+            throw new IllegalStateException("MERGEs nested into the JOINs are not supported");
+        }
+    }
+
+    private boolean hasMergeNestedInJoin() {
+        for (Query<?> joinQuery : joinQueries) {
+            if (!joinQuery.mergeQueries.isEmpty() || joinQuery.hasMergeNestedInJoin()) {
+                return true;
+            }
+        }
+        for (Query<?> mergeQuery : mergeQueries) {
+            if (mergeQuery.hasMergeNestedInJoin()) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     /**
      * Returns the string representation of the query.
      *

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

@@ -94,12 +94,7 @@ private void parseQueryResult(QueryResult queryResult) {
             if (queryResult.isJson()) {
                 throw new UnsupportedOperationException("Query result in json format is not supported");
             } else {
-                CtagMatcher ctagMatcher = new CtagMatcher();
-                PayloadType payloadType = namespace.getPayloadType();
-                if (payloadType != null) {
-                    ctagMatcher.read(payloadType);
-                }
-                itemReader = new CjsonItemReader<>(itemClass, ctagMatcher);
+                itemReader = newItemReader(itemClass, namespace);
             }
         }
     }
@@ -128,22 +123,15 @@ public T next() {
             fetchResults();
         }
 
-        T item = itemClass.cast(readItem(namespace, itemReader, query));
+        T item = itemClass.cast(readItem(itemReader, query));
         position++;
         return item;
-
     }
 
-    private <S> S readItem(ReindexerNamespace<?> expectedNamespace, ItemReader<S> reader, Query<?> queryContext) {
+    private <S> S readItem(ItemReader<S> reader, Query<?> queryContext) {
         ItemParams params = readItemParams();
         Query<?> itemQueryContext = getItemQueryContext(queryContext, params.nsId);
-
-        ReindexerNamespace<?> itemNamespace = expectedNamespace;
-        if (query != null && params.nsId < query.getNamespaces().size()) {
-            itemNamespace = query.getNamespaces().get(params.nsId);
-        }
-
-        S item = readItemData(params, reader, itemNamespace);
+        S item = readItemData(params, reader);
         readJoinedItems(item, itemQueryContext, params.nsId);
         return item;
     }
@@ -161,7 +149,7 @@ private Query<?> getItemQueryContext(Query<?> defaultQueryContext, int nsId) {
         return defaultQueryContext;
     }
 
-    private <S> S readItemData(ItemParams params, ItemReader<S> reader, ReindexerNamespace<?> itemNamespace) {
+    private <S> S readItemData(ItemParams params, ItemReader<S> reader) {
         if (params.cptr != 0) {
             ByteBuffer nativeBuffer = NativeUtils.getNativeBuffer(queryResult.getResultsPtr(), params.cptr,
                     params.nsId);
@@ -194,10 +182,10 @@ private void readJoinedItems(Object item, Query<?> queryContext, int nsId) {
 
             Query<?> joinQuery = queryContext.getJoinQueries().get(joinedField);
             ReindexerNamespace<?> joinedNamespace = joinQuery.getNamespace();
-            CjsonItemReader<?> joinedItemReader = newItemReader(joinedNamespace);
+            CjsonItemReader<?> joinedItemReader = newItemReader(joinedNamespace.getItemClass(), joinedNamespace);
             List<Object> subItems = new ArrayList<>(itemsCount);
             for (int i = 0; i < itemsCount; i++) {
-                subItems.add(readItem(joinedNamespace, joinedItemReader, joinQuery));
+                subItems.add(readItem(joinedItemReader, joinQuery));
             }
             subItemsMap.computeIfAbsent(queryContext.getJoinFields().get(joinedField), field -> new ArrayList<>())
                     .addAll(subItems);
@@ -219,11 +207,11 @@ private void readJoinedItemsV1(Object item, int nsId) {
         for (int nsIndex = 0; nsIndex < joinedFields; nsIndex++) {
             int itemsCount = (int) buffer.getVarUInt();
             ReindexerNamespace<?> joinedNamespace = query.getNamespaces().get(nsIndex + namespaceIndexOffset);
-            CjsonItemReader<?> joinedItemReader = newItemReader(joinedNamespace);
+            CjsonItemReader<?> joinedItemReader = newItemReader(joinedNamespace.getItemClass(), joinedNamespace);
             List<Object> subItems = new ArrayList<>(itemsCount);
             for (int j = 0; j < itemsCount; j++) {
                 ItemParams subItemParams = readItemParams();
-                subItems.add(readItemData(subItemParams, joinedItemReader, joinedNamespace));
+                subItems.add(readItemData(subItemParams, joinedItemReader));
             }
 
             String joinField = query.getJoinFields().get(nsIndex);
@@ -281,11 +269,13 @@ private int getJoinedNsIndexOffset(int nsId) {
         return offset;
     }
 
-    private CjsonItemReader<?> newItemReader(ReindexerNamespace<?> itemNamespace) {
-        PayloadType payloadType = itemNamespace.getPayloadType();
+    private <S> CjsonItemReader<S> newItemReader(Class<S> itemClass, ReindexerNamespace<?> itemNamespace) {
         CtagMatcher ctagMatcher = new CtagMatcher();
-        ctagMatcher.read(payloadType);
-        return new CjsonItemReader<>(itemNamespace.getItemClass(), ctagMatcher);
+        PayloadType payloadType = itemNamespace.getPayloadType();
+        if (payloadType != null) {
+            ctagMatcher.read(payloadType);
+        }
+        return new CjsonItemReader<>(itemClass, ctagMatcher);
     }
 
     private void writeJoinResult(Object item, String fieldName, List<Object> subItems) {

src/test/java/ru/rt/restream/reindexer/connector/BuiltinNestedJoinTest.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020-present Restream
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ru.rt.restream.reindexer.connector;
+
+import ru.rt.restream.category.BuiltinTest;
+
+/**
+ * Tests for Builtin implementation.
+ */
+@BuiltinTest
+public class BuiltinNestedJoinTest extends NestedJoinTest {
+
+}

src/test/java/ru/rt/restream/reindexer/connector/CprotoNestedJoinTest.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020-present Restream
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ru.rt.restream.reindexer.connector;
+
+import ru.rt.restream.category.CprotoTest;
+
+/**
+ * Tests for Cproto implementation.
+ */
+@CprotoTest
+public class CprotoNestedJoinTest extends NestedJoinTest {
+
+}