DATACOUCH-388 Add support for ANSI-JOIN across entities

Motivation
----------
To support ANSI JOIN across associated entities

Changes
-------
N1qlJoin annotation on an associated entity field is discovered by the
couchbaseTemplate, which uses N1qlJoinResolver to build and resolve the
query. The query can be resolved eagerly or lazily based on the fetch
type configuration. The retrieved results are then mapped to the
associated entity. In the lazy resolver, a proxy is set on the property
which resolves on the first access.

Results
-------
Verified by unit and integration tests. ANSI join is now possible across
entities.

Original PR: #174

src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java

@@ -50,12 +50,16 @@ import com.couchbase.client.java.view.ViewQuery;
 import com.couchbase.client.java.view.ViewResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.convert.join.N1qlJoinResolver;
 import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
 import org.springframework.data.couchbase.core.mapping.CouchbaseMappingContext;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
 import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
 import org.springframework.data.couchbase.core.mapping.CouchbaseStorable;
 import org.springframework.data.couchbase.core.mapping.KeySettings;
+import org.springframework.data.couchbase.core.query.N1qlJoin;
+import org.springframework.data.mapping.PropertyHandler;
+import org.springframework.data.util.TypeInformation;
 import rx.Observable;
 import rx.functions.Func1;
 
@@ -714,6 +718,22 @@ public class CouchbaseTemplate implements CouchbaseOperations, ApplicationEventP
       accessor.setProperty(persistentEntity.getVersionProperty(), data.cas());
     }
 
+    persistentEntity.doWithProperties((PropertyHandler<CouchbasePersistentProperty>) prop -> {
+      if (prop.isAnnotationPresent(N1qlJoin.class)) {
+        N1qlJoin definition = prop.findAnnotation(N1qlJoin.class);
+        TypeInformation type = prop.getTypeInformation().getComponentType();
+        Class clazz = type.getType();
+        N1qlJoinResolver.N1qlJoinResolverParameters parameters = new N1qlJoinResolver.N1qlJoinResolverParameters(definition, id, persistentEntity.getTypeInformation(), type);
+        if (N1qlJoinResolver.isLazyJoin(definition)) {
+          N1qlJoinResolver.N1qlJoinProxy proxy = new N1qlJoinResolver.N1qlJoinProxy(this, parameters);
+          accessor.setProperty(prop, java.lang.reflect.Proxy.newProxyInstance(List.class.getClassLoader(),
+                  new Class[]{List.class}, proxy));
+        } else {
+          accessor.setProperty(prop, N1qlJoinResolver.doResolve(this, parameters, clazz));
+        }
+      }
+    });
+
     return accessor.getBean();
   }
 

src/main/java/org/springframework/data/couchbase/core/convert/MappingCouchbaseConverter.java

@@ -43,6 +43,7 @@ import org.springframework.data.couchbase.core.mapping.id.GeneratedValue;
 import org.springframework.data.couchbase.core.mapping.id.IdAttribute;
 import org.springframework.data.couchbase.core.mapping.id.IdPrefix;
 import org.springframework.data.couchbase.core.mapping.id.IdSuffix;
+import org.springframework.data.couchbase.core.query.N1qlJoin;
 import org.springframework.data.mapping.Association;
 import org.springframework.data.mapping.AssociationHandler;
 import org.springframework.data.mapping.MappingException;
@@ -239,7 +240,7 @@ public class MappingCouchbaseConverter extends AbstractCouchbaseConverter
     entity.doWithProperties(new PropertyHandler<CouchbasePersistentProperty>() {
       @Override
       public void doWithPersistentProperty(final CouchbasePersistentProperty prop) {
-        if (!doesPropertyExistInSource(prop) || entity.isConstructorArgument(prop) || isIdConstructionProperty(prop)) {
+        if (!doesPropertyExistInSource(prop) || entity.isConstructorArgument(prop) || isIdConstructionProperty(prop) || prop.isAnnotationPresent(N1qlJoin.class)) {
           return;
         }
         Object obj = prop.isIdProperty() ? source.getId() : getValueInternal(prop, source, instance);
@@ -482,6 +483,8 @@ public class MappingCouchbaseConverter extends AbstractCouchbaseConverter
           return;
         } else if (enableStrictFieldChecking && !prop.isAnnotationPresent(Field.class)) {
           return;
+        } else if (prop.isAnnotationPresent(N1qlJoin.class)) {
+          return;
         }
 
         Object propertyObj = accessor.getProperty(prop, prop.getType());

src/main/java/org/springframework/data/couchbase/core/convert/join/N1qlJoinResolver.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2018 the original author or authors
+ *
+ * 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 org.springframework.data.couchbase.core.convert.join;
+
+import static org.springframework.data.couchbase.core.support.TemplateUtils.*;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.util.List;
+
+import com.couchbase.client.java.query.N1qlQuery;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.CouchbaseTemplate;
+import org.springframework.data.couchbase.core.query.FetchType;
+import org.springframework.data.couchbase.core.query.HashSide;
+import org.springframework.data.couchbase.core.query.N1qlJoin;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.util.Assert;
+
+/**
+ * N1qlJoinResolver resolves by converting the join definition to query statement
+ * and executing using CouchbaseTemplate
+ *
+ * @author Subhashni Balakrishnan
+ */
+public class N1qlJoinResolver {
+    private static final Logger LOGGER = LoggerFactory.getLogger(N1qlJoinResolver.class);
+
+    public static String buildQuery(CouchbaseTemplate template, N1qlJoinResolverParameters parameters) {
+        String joinType = "JOIN";
+        String selectEntity = "SELECT META(rks).id AS " + SELECT_ID +
+                ", META(rks).cas AS " + SELECT_CAS + ", (rks).* ";
+
+        StringBuilder useLKSBuilder = new StringBuilder();
+        if (parameters.getJoinDefinition().index().length() > 0) {
+            useLKSBuilder.append("INDEX(" + parameters.getJoinDefinition().index() + ")");
+        }
+        String useLKS = useLKSBuilder.length() > 0 ? "USE " + useLKSBuilder.toString() + " " : "";
+
+        String from = "FROM `" + template.getCouchbaseBucket().name() + "` lks " + useLKS + joinType + " " + template.getCouchbaseBucket().name() + " rks";
+        String onLks = "lks." + template.getConverter().getTypeKey() + " = \""+ parameters.getEntityTypeInfo().getType().getName() + "\"";
+        String onRks = "rks." + template.getConverter().getTypeKey() + " = \"" + parameters.getAssociatedEntityTypeInfo().getType().getName() + "\"";
+
+
+        StringBuilder useRKSBuilder = new StringBuilder();
+        if (parameters.getJoinDefinition().rightIndex().length() > 0) {
+            useRKSBuilder.append("INDEX(" + parameters.getJoinDefinition().rightIndex() + ")");
+        }
+        if (!parameters.getJoinDefinition().hashside().equals(HashSide.NONE)) {
+            if (useRKSBuilder.length() > 0) useRKSBuilder.append(" ");
+            useRKSBuilder.append("HASH(" + parameters.getJoinDefinition().hashside().getValue() +")");
+        }
+        if (parameters.getJoinDefinition().keys().length > 0) {
+            if (useRKSBuilder.length() > 0) useRKSBuilder.append(" ");
+            useRKSBuilder.append("KEYS [");
+            String[] keys = parameters.getJoinDefinition().keys();
+
+            for(int i=0; i < keys.length;i++) {
+                if(i != 0) useRKSBuilder.append(",");
+                useRKSBuilder.append("\"" + keys[i] +"\"");
+            }
+            useRKSBuilder.append("]");
+        }
+
+        String on = "ON " + parameters.getJoinDefinition().on().concat(" AND " + onLks).concat(" AND " + onRks);
+
+        String where = "WHERE META(lks).id=\"" + parameters.getLksId() + "\"";
+        where += ((parameters.getJoinDefinition().where().length() > 0) ? " AND " + parameters.getJoinDefinition().where() : "");
+
+        StringBuilder statementSb = new StringBuilder();
+        statementSb.append(selectEntity);
+        statementSb.append(" " + from);
+        statementSb.append((useRKSBuilder.length() > 0? " USE "+ useRKSBuilder.toString() : ""));
+        statementSb.append(" " + on);
+        statementSb.append(" " + where);
+        return statementSb.toString();
+    }
+
+    public static <R> List<R> doResolve(CouchbaseTemplate template,
+                                        N1qlJoinResolverParameters parameters,
+                                        Class<R> associatedEntityClass) {
+        String statement = buildQuery(template, parameters);
+
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug("Join query executed " + statement);
+        }
+
+        N1qlQuery query = N1qlQuery.simple(statement);
+        return template.findByN1QL(query, associatedEntityClass);
+    }
+
+    public static boolean isLazyJoin(N1qlJoin joinDefinition) {
+        return joinDefinition.fetchType().equals(FetchType.LAZY);
+    }
+
+    static public class N1qlJoinProxy implements InvocationHandler {
+        private final CouchbaseTemplate template;
+        private final N1qlJoinResolverParameters params;
+        private List<?> resolved = null;
+
+        public N1qlJoinProxy(CouchbaseTemplate template, N1qlJoinResolverParameters params) {
+            this.template = template;
+            this.params = params;
+        }
+
+        @Override
+        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+            if(this.resolved == null) {
+                this.resolved = doResolve(this.template, this.params, this.params.associatedEntityTypeInfo.getType());
+            }
+            return method.invoke(this.resolved, args);
+        }
+    }
+
+    static public class N1qlJoinResolverParameters {
+        private N1qlJoin joinDefinition;
+        private String lksId;
+        private TypeInformation<?> entityTypeInfo;
+        private TypeInformation<?> associatedEntityTypeInfo;
+
+        public N1qlJoinResolverParameters(N1qlJoin joinDefinition,
+                                          String lksId,
+                                          TypeInformation<?> entityTypeInfo,
+                                          TypeInformation<?> associatedEntityTypeInfo) {
+            Assert.notNull(joinDefinition, "The join definition is required");
+            Assert.notNull(entityTypeInfo, "The entity type information is required");
+            Assert.notNull(associatedEntityTypeInfo, "The associated entity type information is required");
+
+            this.joinDefinition = joinDefinition;
+            this.lksId = lksId;
+            this.entityTypeInfo = entityTypeInfo;
+            this.associatedEntityTypeInfo = associatedEntityTypeInfo;
+        }
+
+        public N1qlJoin getJoinDefinition() {
+            return joinDefinition;
+        }
+
+        public String getLksId() {
+            return lksId;
+        }
+
+        public TypeInformation getEntityTypeInfo() {
+            return entityTypeInfo;
+        }
+
+        public TypeInformation getAssociatedEntityTypeInfo() {
+            return associatedEntityTypeInfo;
+        }
+    }
+}

src/main/java/org/springframework/data/couchbase/core/query/FetchType.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2018 the original author or authors
+ *
+ * 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 org.springframework.data.couchbase.core.query;
+
+/**
+ * Setting for specify when to fetch the associated entities
+ * @author Subhashni Balakrishnan
+ */
+public enum FetchType {
+	/**
+	 * Immediately fetch the associated entities
+	 */
+	IMMEDIATE,
+
+	/**
+	 * Lazily fetch the associated entities on access, the
+	 * fetch happens only once
+	 */
+	LAZY
+}

src/main/java/org/springframework/data/couchbase/core/query/HashSide.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2018 the original author or authors
+ *
+ * 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 org.springframework.data.couchbase.core.query;
+
+/**
+ * Hash side to specify hash join. Here based on probe or build, the
+ * entity will be used to query or build the hash table.
+ * The smaller data set side should be used to build to fit in memory.
+ *
+ * @author Subhashni Balakrishnan
+ */
+public enum HashSide {
+	/**
+	 * Hash join will not be used
+	 */
+	NONE("none"),
+
+	/**
+	 * Associated entity will be on the probe side of the hash table
+	 */
+	PROBE("probe"),
+
+	/**
+	 * Associated entity will be used to build the hash table for faster lookup
+	 */
+	BUILD("build");
+
+	private final String value;
+
+	HashSide(String value) {
+		this.value=value;
+	}
+
+	public String getValue() {
+		return this.value;
+	}
+}

src/main/java/org/springframework/data/couchbase/core/query/N1qlJoin.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2018 the original author or authors
+ *
+ * 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 org.springframework.data.couchbase.core.query;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * This annotation is targeted for entity field which is a list of the
+ * associated entities fetched by ANSI Join across the entities available
+ * from Couchbase Server 5.5
+ *
+ * @author Subhashni Balakrishnan
+ */
+@Target({ElementType.FIELD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface N1qlJoin {
+    /**
+     * Join Criteria can be a simple equi join or multiple conditions
+     * combined using AND or OR. Array based equi joins with unnest is
+     * also possible. To reference fields in entity use prefix "lks."
+     * (left key space) and for referencing fields in associated entities
+     * use "rks." (right key space)
+     */
+    String on();
+
+    /**
+     * Fetch type specifies how the associated entities are fetched
+     * {@link FetchType}
+     */
+    FetchType fetchType() default FetchType.IMMEDIATE;
+
+    /**
+     * Where clause for the join. To reference fields in entity use
+     * prefix "lks." and for referencing fields in associated entities
+     * use "rks."
+     */
+    String where() default "";
+
+    /**
+     * Hint index for entity for indexed nested loop join
+     */
+    String index() default "";
+
+    /**
+     * Hint index for associated entity for indexed nested loop join
+     */
+    String rightIndex() default "";
+
+    /**
+     * Hash side specification for the associated entity for hash join
+     * Note: Supported on enterprise edition only
+     */
+    HashSide hashside() default HashSide.NONE;
+
+    /**
+     * Use keys query hint
+     */
+    String[] keys() default {};
+}