ATLAS-1369 : Optimize Gremlin queries generated by DSL translator

common/src/main/java/org/apache/atlas/groovy/AbstractFunctionExpression.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.groovy;
+
+/**
+ * Base class for all expression that can have a caller.
+ */
+public abstract class AbstractFunctionExpression extends AbstractGroovyExpression {
+
+    // null for global functions
+    private GroovyExpression caller;
+    private TraversalStepType type = TraversalStepType.NONE;
+
+    public AbstractFunctionExpression(GroovyExpression target) {
+        this.caller = target;
+    }
+
+    public AbstractFunctionExpression(TraversalStepType type, GroovyExpression target) {
+        this.caller = target;
+        this.type = type;
+    }
+
+    public  GroovyExpression getCaller() {
+        return caller;
+    }
+
+    public void setCaller(GroovyExpression expr) {
+        caller = expr;
+    }
+
+
+    public void setType(TraversalStepType type) {
+        this.type = type;
+    }
+
+    @Override
+    public TraversalStepType getType() {
+        return type;
+    }
+
+
+}

common/src/main/java/org/apache/atlas/groovy/AbstractGroovyExpression.java

@@ -33,4 +33,13 @@ public abstract class AbstractGroovyExpression implements GroovyExpression {
         return ctx.getQuery();
     }
 
+    @Override
+    public TraversalStepType getType() {
+        return TraversalStepType.NONE;
+    }
+
+    @Override
+    public GroovyExpression copy() {
+        return copy(getChildren());
+    }
 }

common/src/main/java/org/apache/atlas/groovy/ArithmeticExpression.java

@@ -17,6 +17,8 @@
  */
 package org.apache.atlas.groovy;
 
+import java.util.List;
+
 import org.apache.atlas.AtlasException;
 
 /**
@@ -56,4 +58,14 @@ public class ArithmeticExpression extends BinaryExpression {
     public ArithmeticExpression(GroovyExpression left, ArithmeticOperator op, GroovyExpression right) {
         super(left, op.getGroovyValue(), right);
     }
+
+    private ArithmeticExpression(GroovyExpression left, String op, GroovyExpression right) {
+        super(left, op, right);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 2;
+        return new ArithmeticExpression(newChildren.get(0), op, newChildren.get(1));
+    }
 }

common/src/main/java/org/apache/atlas/groovy/BinaryExpression.java

@@ -18,6 +18,9 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Arrays;
+import java.util.List;
+
 /**
  * Represents any kind of binary expression.  This could
  * be an arithmetic expression, such as a + 3, a boolean
@@ -30,7 +33,7 @@ public abstract class BinaryExpression extends AbstractGroovyExpression {
 
     private GroovyExpression left;
     private GroovyExpression right;
-    private String op;
+    protected String op;
 
     public BinaryExpression(GroovyExpression left, String op, GroovyExpression right) {
         this.left = left;
@@ -48,4 +51,8 @@ public abstract class BinaryExpression extends AbstractGroovyExpression {
         right.generateGroovy(context);
     }
 
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Arrays.asList(left, right);
+    }
 }

common/src/main/java/org/apache/atlas/groovy/CastExpression.java

@@ -18,6 +18,9 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Collections;
+import java.util.List;
+
 /**
  * Groovy expression that represents a cast.
  */
@@ -28,7 +31,7 @@ public class CastExpression extends AbstractGroovyExpression {
 
     public CastExpression(GroovyExpression expr, String className) {
         this.expr = expr;
-        this.className  =className;
+        this.className  = className;
     }
 
     @Override
@@ -41,4 +44,13 @@ public class CastExpression extends AbstractGroovyExpression {
         context.append(")");
     }
 
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.singletonList(expr);
+    }
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 1;
+        return new CastExpression(newChildren.get(0), className);
+    }
 }

common/src/main/java/org/apache/atlas/groovy/ClosureExpression.java

@@ -20,6 +20,7 @@ package org.apache.atlas.groovy;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -28,28 +29,79 @@ import java.util.List;
  */
 public class ClosureExpression extends AbstractGroovyExpression {
 
-    private List<String> varNames = new ArrayList<>();
-    private GroovyExpression body;
+    /**
+     * Variable declaration in a closure.
+     */
+    public static class VariableDeclaration {
+        private String type;
+        private String varName;
+
+        public VariableDeclaration(String type, String varName) {
+            super();
+            this.type = type;
+            this.varName = varName;
+        }
+
+        public VariableDeclaration(String varName) {
+            this.varName = varName;
+        }
+
+        public void append(GroovyGenerationContext context) {
+            if (type != null) {
+                context.append(type);
+                context.append(" ");
+            }
+            context.append(varName);
+        }
+    }
+    private List<VariableDeclaration> vars = new ArrayList<>();
+    private StatementListExpression body = new StatementListExpression();
+
+    public ClosureExpression(String... varNames) {
+        this(null, varNames);
+    }
 
-    public ClosureExpression(GroovyExpression body, String... varNames) {
-        this.body = body;
-        this.varNames.addAll(Arrays.asList(varNames));
+    public ClosureExpression(GroovyExpression initialStmt, String... varNames) {
+        this(Arrays.asList(varNames), initialStmt);
     }
 
-    public ClosureExpression(List<String> varNames, GroovyExpression body) {
-        this.body = body;
-        this.varNames.addAll(varNames);
+    public ClosureExpression(List<String> varNames, GroovyExpression initialStmt) {
+        if (initialStmt != null) {
+            this.body.addStatement(initialStmt);
+        }
+        for (String varName : varNames) {
+            vars.add(new VariableDeclaration(varName));
+        }
+    }
+
+    public ClosureExpression(GroovyExpression initialStmt, List<VariableDeclaration> varNames) {
+        if (initialStmt != null) {
+            this.body.addStatement(initialStmt);
+        }
+        vars.addAll(varNames);
+    }
+
+    public void addStatement(GroovyExpression expr) {
+        body.addStatement(expr);
+    }
+
+    public void addStatements(List<GroovyExpression> exprs) {
+        body.addStatements(exprs);
+    }
+
+    public void replaceStatement(int index, GroovyExpression newExpr) {
+        body.replaceStatement(index, newExpr);
     }
 
     @Override
     public void generateGroovy(GroovyGenerationContext context) {
 
         context.append("{");
-        if (!varNames.isEmpty()) {
-            Iterator<String> varIt = varNames.iterator();
+        if (!vars.isEmpty()) {
+            Iterator<VariableDeclaration> varIt = vars.iterator();
             while(varIt.hasNext()) {
-                String varName = varIt.next();
-                context.append(varName);
+                VariableDeclaration var = varIt.next();
+                var.append(context);
                 if (varIt.hasNext()) {
                     context.append(", ");
                 }
@@ -58,6 +110,20 @@ public class ClosureExpression extends AbstractGroovyExpression {
         }
         body.generateGroovy(context);
         context.append("}");
+    }
+
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.<GroovyExpression>singletonList(body);
+    }
 
+    public List<GroovyExpression> getStatements() {
+        return body.getStatements();
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 1;
+        return new ClosureExpression(newChildren.get(0), vars);
     }
 }

common/src/main/java/org/apache/atlas/groovy/ComparisonExpression.java

@@ -17,6 +17,8 @@
  */
 package org.apache.atlas.groovy;
 
+import java.util.List;
+
 import org.apache.atlas.AtlasException;
 
 /**
@@ -61,4 +63,14 @@ public class ComparisonExpression extends BinaryExpression {
     public ComparisonExpression(GroovyExpression left, ComparisonOperator op, GroovyExpression right) {
         super(left, op.getGroovyValue(), right);
     }
+
+    private ComparisonExpression(GroovyExpression left, String op, GroovyExpression right) {
+        super(left, op, right);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 2;
+        return new ComparisonExpression(newChildren.get(0), op, newChildren.get(1));
+    }
 }

common/src/main/java/org/apache/atlas/groovy/ComparisonOperatorExpression.java

@@ -17,6 +17,8 @@
  */
 package org.apache.atlas.groovy;
 
+import java.util.List;
+
 /**
  * Represents an expression that compares two expressions using
  * the Groovy "spaceship" operator.  This is basically the
@@ -29,4 +31,10 @@ public class ComparisonOperatorExpression extends BinaryExpression {
     public ComparisonOperatorExpression(GroovyExpression left, GroovyExpression right) {
         super(left, "<=>", right);
     }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 2;
+        return new ComparisonOperatorExpression(newChildren.get(0), newChildren.get(1));
+    }
 }

common/src/main/java/org/apache/atlas/groovy/FieldExpression.java

@@ -18,27 +18,38 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Collections;
+import java.util.List;
+
 /**
  * Groovy expression that accesses a field in an object.
  */
-public class FieldExpression extends AbstractGroovyExpression {
+public class FieldExpression extends AbstractFunctionExpression {
 
-    private GroovyExpression target;
     private String fieldName;
 
     public FieldExpression(GroovyExpression target, String fieldName) {
-        this.target = target;
+        super(target);
         this.fieldName = fieldName;
     }
 
     @Override
     public void generateGroovy(GroovyGenerationContext context) {
-
-        target.generateGroovy(context);
+        getCaller().generateGroovy(context);
         context.append(".'");
 
         context.append(fieldName);
         context.append("'");
     }
 
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.singletonList(getCaller());
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 1;
+        return new FieldExpression(newChildren.get(0), fieldName);
+    }
 }

common/src/main/java/org/apache/atlas/groovy/FunctionCallExpression.java

@@ -20,41 +20,52 @@ package org.apache.atlas.groovy;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
 /**
  * Groovy expression that calls a method on an object.
  */
-public class FunctionCallExpression extends AbstractGroovyExpression {
-
-    // null for global functions
-    private GroovyExpression target;
+public class FunctionCallExpression extends AbstractFunctionExpression {
 
     private String functionName;
     private List<GroovyExpression> arguments = new ArrayList<>();
 
-    public FunctionCallExpression(String functionName, List<? extends GroovyExpression> arguments) {
-        this.target = null;
+    public FunctionCallExpression(TraversalStepType type, String functionName, GroovyExpression... arguments) {
+        super(type, null);
         this.functionName = functionName;
-        this.arguments.addAll(arguments);
+        this.arguments.addAll(Arrays.asList(arguments));
     }
 
-    public FunctionCallExpression(GroovyExpression target, String functionName,
-            List<? extends GroovyExpression> arguments) {
-        this.target = target;
+    public FunctionCallExpression(String functionName, GroovyExpression... arguments) {
+        super(null);
+        this.functionName = functionName;
+        this.arguments.addAll(Arrays.asList(arguments));
+    }
+
+    public FunctionCallExpression(TraversalStepType type, String functionName, List<GroovyExpression> arguments) {
+        super(type, null);
         this.functionName = functionName;
         this.arguments.addAll(arguments);
     }
 
-    public FunctionCallExpression(String functionName, GroovyExpression... arguments) {
-        this.target = null;
+    public FunctionCallExpression(GroovyExpression target, String functionName, GroovyExpression... arguments) {
+        super(target);
         this.functionName = functionName;
         this.arguments.addAll(Arrays.asList(arguments));
     }
 
-    public FunctionCallExpression(GroovyExpression target, String functionName, GroovyExpression... arguments) {
-        this.target = target;
+    public FunctionCallExpression(TraversalStepType type, GroovyExpression target, String functionName,
+                                  List<? extends GroovyExpression> arguments) {
+        super(type, target);
+        this.functionName = functionName;
+        this.arguments.addAll(arguments);
+    }
+
+    public FunctionCallExpression(TraversalStepType type, GroovyExpression target, String functionName,
+                                  GroovyExpression... arguments) {
+        super(type, target);
         this.functionName = functionName;
         this.arguments.addAll(Arrays.asList(arguments));
     }
@@ -63,11 +74,20 @@ public class FunctionCallExpression extends AbstractGroovyExpression {
         arguments.add(expr);
     }
 
+    public List<GroovyExpression> getArguments() {
+        return Collections.unmodifiableList(arguments);
+    }
+
+
+    public String getFunctionName() {
+        return functionName;
+    }
+
     @Override
     public void generateGroovy(GroovyGenerationContext context) {
 
-        if (target != null) {
-            target.generateGroovy(context);
+        if (getCaller() != null) {
+            getCaller().generateGroovy(context);
             context.append(".");
         }
         context.append(functionName);
@@ -77,10 +97,44 @@ public class FunctionCallExpression extends AbstractGroovyExpression {
             GroovyExpression expr = it.next();
             expr.generateGroovy(context);
             if (it.hasNext()) {
-                context.append(", ");
+                context.append(",");
             }
         }
         context.append(")");
     }
 
+    @Override
+    public List<GroovyExpression> getChildren() {
+        List<GroovyExpression> result = new ArrayList<>(arguments.size() + 1);
+        if (getCaller() != null) {
+            result.add(getCaller());
+        }
+        result.addAll(arguments);
+        return result;
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+
+        if (getCaller() == null) {
+            return new FunctionCallExpression(getType(), functionName, newChildren);
+        }
+
+        GroovyExpression newTarget = newChildren.get(0);
+        List<GroovyExpression> args = null;
+        if (newChildren.size() > 1) {
+            args = newChildren.subList(1, newChildren.size());
+        } else {
+            args = Collections.emptyList();
+        }
+        return new FunctionCallExpression(getType(), newTarget, functionName, args);
+
+    }
+
+    public void setArgument(int index, GroovyExpression value) {
+        if (index < 0 || index >= arguments.size()) {
+            throw new IllegalArgumentException("Invalid argIndex " + index);
+        }
+        arguments.set(index, value);
+    }
 }

common/src/main/java/org/apache/atlas/groovy/GroovyExpression.java

@@ -18,17 +18,55 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.List;
+
 /**
  * Represents an expression in the Groovy programming language, which
  * is the language that Gremlin scripts are written and interpreted in.
  */
 public interface GroovyExpression  {
-
     /**
-     * Generates a groovy script from the expression.
+     * Generates a Groovy script from the expression.
      *
      * @param context
      */
     void generateGroovy(GroovyGenerationContext context);
 
+    /**
+     * Gets all of the child expressions of this expression.
+     * s
+     * @return
+     */
+    List<GroovyExpression> getChildren();
+
+    /**
+     * Makes a copy of the expression, keeping everything the
+     * same except its child expressions.  These are replaced
+     * with the provided children.  The order of the children
+     * is important.  It is expected that the children provided
+     * here are updated versions of the children returned by
+     * getChildren().  The order of the children must be the
+     * same as the order in which the children were returned
+     * by getChildren()
+     *
+     * @param newChildren
+     * @return
+     */
+    GroovyExpression copy(List<GroovyExpression> newChildren);
+
+    /**
+     * Makes a shallow copy of the GroovyExpression.  This
+     * is equivalent to copy(getChildren());
+     *
+     * @return
+     */
+    GroovyExpression copy();
+
+    /**
+     * Gets the type of traversal step represented by this
+     * expression (or TraversalStepType.NONE if it is not part of a graph traversal).
+     *
+     * @return
+     */
+    TraversalStepType getType();
 }

common/src/main/java/org/apache/atlas/groovy/IdentifierExpression.java

@@ -18,18 +18,28 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Collections;
+import java.util.List;
+
 /**
  * Groovy expression that references the variable with the given name.
  *
  */
 public class IdentifierExpression extends AbstractGroovyExpression {
 
+    private TraversalStepType type = TraversalStepType.NONE;
     private String varName;
 
     public IdentifierExpression(String varName) {
         this.varName = varName;
     }
 
+    public IdentifierExpression(TraversalStepType type, String varName) {
+        this.varName = varName;
+        this.type = type;
+    }
+
+
     public String getVariableName() {
         return varName;
     }
@@ -39,4 +49,25 @@ public class IdentifierExpression extends AbstractGroovyExpression {
         context.append(varName);
     }
 
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.emptyList();
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.isEmpty();
+        IdentifierExpression result =  new IdentifierExpression(varName);
+        result.setType(type);
+        return result;
+    }
+
+    public void setType(TraversalStepType type) {
+        this.type = type;
+    }
+
+    @Override
+    public TraversalStepType getType() {
+        return type;
+    }
 }

common/src/main/java/org/apache/atlas/groovy/LabeledExpression.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.groovy;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Represents a Groovy expression that has a label.
+ */
+public class LabeledExpression extends AbstractGroovyExpression {
+
+    private String label;
+    private GroovyExpression expr;
+
+    public LabeledExpression(String label, GroovyExpression expr) {
+        this.label = label;
+        this.expr = expr;
+    }
+
+    @Override
+    public void generateGroovy(GroovyGenerationContext context) {
+        context.append(label);
+        context.append(":");
+        expr.generateGroovy(context);
+    }
+
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.singletonList(expr);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 1;
+        return new LabeledExpression(label, newChildren.get(0));
+    }
+}

common/src/main/java/org/apache/atlas/groovy/ListExpression.java

@@ -20,6 +20,7 @@ package org.apache.atlas.groovy;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -57,4 +58,15 @@ public class ListExpression extends AbstractGroovyExpression {
         context.append("]");
     }
 
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.unmodifiableList(values);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        return new ListExpression(newChildren);
+    }
+
+
 }

common/src/main/java/org/apache/atlas/groovy/LiteralExpression.java

@@ -18,13 +18,15 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 /**
  * Represents a literal value.
  */
-public class LiteralExpression implements GroovyExpression {
+public class LiteralExpression extends AbstractGroovyExpression {
 
     public static final LiteralExpression TRUE = new LiteralExpression(true);
     public static final LiteralExpression FALSE = new LiteralExpression(false);
@@ -40,6 +42,12 @@ public class LiteralExpression implements GroovyExpression {
         this.addTypeSuffix = addTypeSuffix;
     }
 
+    public LiteralExpression(Object value, boolean addTypeSuffix, boolean translateToParameter) {
+        this.value = value;
+        this.translateToParameter = translateToParameter;
+        this.addTypeSuffix = addTypeSuffix;
+    }
+
     public LiteralExpression(Object value) {
         this.value = value;
         this.translateToParameter = value instanceof String;
@@ -86,6 +94,10 @@ public class LiteralExpression implements GroovyExpression {
 
     }
 
+    public Object getValue() {
+        return value;
+    }
+
     private String getEscapedValue() {
         String escapedValue = (String)value;
         escapedValue = escapedValue.replaceAll(Pattern.quote("\\"), Matcher.quoteReplacement("\\\\"));
@@ -96,4 +108,15 @@ public class LiteralExpression implements GroovyExpression {
     public void setTranslateToParameter(boolean translateToParameter) {
         this.translateToParameter = translateToParameter;
     }
+
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.emptyList();
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 0;
+        return new LiteralExpression(value, addTypeSuffix, translateToParameter);
+    }
 }

common/src/main/java/org/apache/atlas/groovy/LogicalExpression.java

@@ -17,6 +17,8 @@
  */
 package org.apache.atlas.groovy;
 
+import java.util.List;
+
 /**
  * Represents a logical (and/or) expression.
  *
@@ -43,4 +45,14 @@ public class LogicalExpression extends BinaryExpression {
     public LogicalExpression(GroovyExpression left, LogicalOperator op, GroovyExpression right) {
         super(left, op.getGroovyValue(), right);
     }
+
+    private LogicalExpression(GroovyExpression left, String op, GroovyExpression right) {
+        super(left, op, right);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 2;
+        return new LogicalExpression(newChildren.get(0), op, newChildren.get(1));
+    }
 }

common/src/main/java/org/apache/atlas/groovy/RangeExpression.java

@@ -18,28 +18,68 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Collections;
+import java.util.List;
+
 /**
  * Represents an "exclusive" range expression, e.g. [0..<10].
  */
-public class RangeExpression extends AbstractGroovyExpression {
+public class RangeExpression extends AbstractFunctionExpression {
 
-    private GroovyExpression parent;
-    private int offset;
-    private int count;
+    private TraversalStepType stepType;
+    private int startIndex;
+    private int endIndex;
 
-    public RangeExpression(GroovyExpression parent, int offset, int count) {
-        this.parent = parent;
-        this.offset = offset;
-        this.count = count;
+    public RangeExpression(TraversalStepType stepType, GroovyExpression parent, int offset, int count) {
+        super(parent);
+        this.startIndex = offset;
+        this.endIndex = count;
+        this.stepType = stepType;
     }
 
     @Override
     public void generateGroovy(GroovyGenerationContext context) {
-        parent.generateGroovy(context);
+        getCaller().generateGroovy(context);
         context.append(" [");
-        new LiteralExpression(offset).generateGroovy(context);
+        new LiteralExpression(startIndex).generateGroovy(context);
         context.append("..<");
-        new LiteralExpression(count).generateGroovy(context);
+        new LiteralExpression(endIndex).generateGroovy(context);
         context.append("]");
     }
+
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.singletonList(getCaller());
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 1;
+        return new RangeExpression(stepType, newChildren.get(0), startIndex, endIndex);
+    }
+
+    @Override
+    public TraversalStepType getType() {
+        return stepType;
+    }
+
+    public int getStartIndex() {
+
+        return startIndex;
+    }
+
+    public void setStartIndex(int startIndex) {
+
+        this.startIndex = startIndex;
+    }
+
+    public int getEndIndex() {
+
+        return endIndex;
+    }
+
+    public void setEndIndex(int endIndex) {
+
+        this.endIndex = endIndex;
+    }
 }

common/src/main/java/org/apache/atlas/groovy/CodeBlockExpression.java → common/src/main/java/org/apache/atlas/groovy/StatementListExpression.java

@@ -19,35 +19,46 @@
 package org.apache.atlas.groovy;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
 /**
- * Groovy expression that represents a block of code
- * that contains 0 or more statements that are delimited
- * by semicolons.
+ * Represents a semi-colon delimited list of Groovy expressions.
  */
-public class CodeBlockExpression extends AbstractGroovyExpression {
+public class StatementListExpression extends AbstractGroovyExpression {
 
-    private List<GroovyExpression> body = new ArrayList<>();
+    private List<GroovyExpression> stmts = new ArrayList<>();
+
+    public StatementListExpression() {
+
+    }
+
+    /**
+     * @param newChildren
+     */
+    public StatementListExpression(List<GroovyExpression> newChildren) {
+        stmts.addAll(newChildren);
+    }
 
     public void addStatement(GroovyExpression expr) {
-        body.add(expr);
+        if (expr instanceof StatementListExpression) {
+            stmts.addAll(((StatementListExpression)expr).getStatements());
+        } else {
+            stmts.add(expr);
+        }
     }
 
     public void addStatements(List<GroovyExpression> exprs) {
-        body.addAll(exprs);
+        for(GroovyExpression expr : exprs) {
+            addStatement(expr);
+        }
     }
 
     @Override
     public void generateGroovy(GroovyGenerationContext context) {
 
-        /*
-         * the L:{} represents a groovy code block; the label is needed
-         * to distinguish it from a groovy closure.
-         */
-        context.append("L:{");
-        Iterator<GroovyExpression> stmtIt = body.iterator();
+        Iterator<GroovyExpression> stmtIt = stmts.iterator();
         while(stmtIt.hasNext()) {
             GroovyExpression stmt = stmtIt.next();
             stmt.generateGroovy(context);
@@ -55,7 +66,33 @@ public class CodeBlockExpression extends AbstractGroovyExpression {
                 context.append(";");
             }
         }
-        context.append("}");
+    }
+
 
+    public List<GroovyExpression> getStatements() {
+        return stmts;
+    }
+
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.unmodifiableList(stmts);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        return new StatementListExpression(newChildren);
+    }
+
+    @Override
+    public TraversalStepType getType() {
+        return TraversalStepType.NONE;
+    }
+
+    /**
+     * @param oldExpr
+     * @param newExpr
+     */
+    public void replaceStatement(int index, GroovyExpression newExpr) {
+        stmts.set(index, newExpr);
     }
 }

common/src/main/java/org/apache/atlas/groovy/TernaryOperatorExpression.java

@@ -18,6 +18,9 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Arrays;
+import java.util.List;
+
 /**
  * Groovy expression that represents the ternary operator (expr ? trueValue :
  * falseValue)
@@ -41,9 +44,9 @@ public class TernaryOperatorExpression extends AbstractGroovyExpression {
 
         context.append("((");
         booleanExpr.generateGroovy(context);
-        context.append(") ? (");
+        context.append(")?(");
         trueValue.generateGroovy(context);
-        context.append(") : (");
+        context.append("):(");
         falseValue.generateGroovy(context);
         context.append("))");
     }
@@ -53,4 +56,20 @@ public class TernaryOperatorExpression extends AbstractGroovyExpression {
         generateGroovy(context);
         return context.getQuery();
     }
+
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Arrays.asList(booleanExpr, trueValue, falseValue);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 3;
+        return new TernaryOperatorExpression(newChildren.get(0), newChildren.get(1), newChildren.get(2));
+    }
+
+    @Override
+    public TraversalStepType getType() {
+        return trueValue.getType();
+    }
 }

common/src/main/java/org/apache/atlas/groovy/TraversalStepType.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.groovy;
+
+/**
+ * Types of graph traversal steps.  These are based on the traversal steps
+ * described in the TinkerPop documentation at
+ * http://tinkerpop.apache.org/docs/current/reference/#graph-traversal-steps.
+ */
+public enum TraversalStepType {
+    /**
+     * Indicates that the expression is not part of a graph traversal.
+     */
+    NONE,
+
+    /**
+     * Indicates that the expression is a
+     * {@link org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource}.
+     * This is not technically a graph traversal step.  This is the expression the traversal is started from ("g").
+     */
+    SOURCE,
+
+    /**
+     * A Start step adds vertices or edges to the traversal.  These include "V", "E", and "inject".
+     */
+    START,
+
+    /**
+     * An End step causes the traversal to be executed.  This includes steps such as "toList", "toSet", and "fill"
+     */
+    END,
+
+    /**
+     * Map steps map the current traverser value to exactly one new value.  These
+     * steps include "map" and "select".  Here, we make a further distinction
+     * based on the type of expression that things are being mapped to.
+     * <p>
+     * MAP_TO_ELEMENT indicates that the traverser value is being mapped
+     * to either a Vertex or an Edge.
+     */
+    MAP_TO_ELEMENT,
+    /**
+     * Map steps map the current traverser value to exactly one new value.  These
+     * steps include "map" and "select".  Here, we make a further distinction
+     * based on the type of expression that things are being mapped to.
+     * <p>
+     * MAP_TO_VALUE indicates that the traverser value is being mapped
+     * to something that is not a Vertex or an Edge.
+     */
+    MAP_TO_VALUE,
+
+    /**
+     * FlatMap steps map the current value of the traverser to an iterator of objects that
+     * are streamed to the next step.  These are steps like "in, "out", "inE", and
+     * so forth which map the current value of the traverser from some vertex or edge
+     * to some other set of vertices or edges that is derived from the original set based
+     * on the structure of the graph.  This also includes "values", which maps a vertex or
+     * edge to the set of values for a given property. Here, we make a further distinction
+     * based on the type of expression that things are being mapped to.
+     * <p>
+     *  FLAT_MAP_TO_ELEMENTS indicates that the traverser value is being mapped
+     * to something that is a Vertex or an Edge (in, out, outE fall in this category).
+     */
+    FLAT_MAP_TO_ELEMENTS,
+
+    /**
+     * FlatMap steps map the current value of the traverser to an iterator of objects that
+     * are streamed to the next step.  These are steps like "in, "out", "inE", and
+     * so forth which map the current value of the traverser from some vertex or edge
+     * to some other set of vertices or edges that is derived from the original set based
+     * on the structure of the graph.  This also includes "values", which maps a vertex or
+     * edge to the set of values for a given property. Here, we make a further distinction
+     * based on the type of expression that things are being mapped to.
+     * <p>
+     *  FLAT_MAP_TO_VALUES indicates that the traverser value is being mapped
+     * to something that not is a Vertex or an Edge (values falls in this category).
+     */
+    FLAT_MAP_TO_VALUES,
+
+    /**
+     * Filter steps filter things out of the traversal.  These include "has", "where",
+     * "and", "or", and "filter".
+     */
+    FILTER,
+
+    /**
+     * Side effect steps do not affect the traverser value, but do something
+     * that affects the state of the traverser.  These include things such as
+     * "enablePath()", "as", and "by".
+     */
+    SIDE_EFFECT,
+
+    /**
+     * Branch steps split the traverser, for example, "repeat", "branch", "choose", and "union".
+     */
+    BRANCH,
+
+    /**
+     * Barrier steps in Gremlin force everything before them to be executed
+     * before moving on to the steps after them.  We also use this to indicate
+     * steps that need to do some aggregation or processing that requires the
+     * full query result to be present in order for the step to work correctly.
+     * This includes "range", "group", and "order", and "cap"
+     */
+    BARRIER,
+}

common/src/main/java/org/apache/atlas/groovy/TypeCoersionExpression.java

@@ -18,6 +18,9 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Collections;
+import java.util.List;
+
 /**
  * Groovy expression that represents a type coersion (e.g obj as Set).
  */
@@ -28,17 +31,29 @@ public class TypeCoersionExpression extends AbstractGroovyExpression {
 
     public TypeCoersionExpression(GroovyExpression expr, String className) {
         this.expr = expr;
-        this.className  =className;
+        this.className  = className;
     }
 
     @Override
     public void generateGroovy(GroovyGenerationContext context) {
 
-        context.append("(");
+        context.append("((");
         expr.generateGroovy(context);
         context.append(")");
         context.append(" as ");
         context.append(className);
+        context.append(")");
+    }
+
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.singletonList(expr);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 1;
+        return new TypeCoersionExpression(newChildren.get(0), className);
     }
 
 }

common/src/main/java/org/apache/atlas/groovy/VariableAssignmentExpression.java

@@ -18,6 +18,9 @@
 
 package org.apache.atlas.groovy;
 
+import java.util.Collections;
+import java.util.List;
+
 /**
  * Groovy statement that assigns a value to a variable.
  */
@@ -50,9 +53,20 @@ public class VariableAssignmentExpression extends AbstractGroovyExpression {
             context.append(" ");
         }
         context.append(name);
-        context.append(" = ");
+        context.append("=");
         value.generateGroovy(context);
 
     }
 
+    @Override
+    public List<GroovyExpression> getChildren() {
+        return Collections.singletonList(value);
+    }
+
+    @Override
+    public GroovyExpression copy(List<GroovyExpression> newChildren) {
+        assert newChildren.size() == 1;
+        return new VariableAssignmentExpression(name, newChildren.get(0));
+    }
+
 }

distro/src/conf/atlas-application.properties

@@ -29,6 +29,13 @@ atlas.graph.storage.hbase.table=apache_atlas_titan
 
 ${titan.storage.properties}
 
+# Gremlin Query Optimizer
+#
+# Enables rewriting gremlin queries to maximize performance. This flag is provided as
+# a possible way to work around any defects that are found in the optimizer until they
+# are resolved.
+#atlas.query.gremlinOptimizerEnabled=true
+
 # Delete handler
 #
 # This allows the default behavior of doing "soft" deletes to be changed.

graphdb/titan0/pom.xml

@@ -34,6 +34,7 @@
     <properties>
         <tinkerpop.version>2.6.0</tinkerpop.version>
         <titan.version>0.5.4</titan.version>
+	<guava.version>14.0</guava.version>
     </properties>
 
     <dependencies>
@@ -53,6 +54,13 @@
         </dependency>
 
         <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${guava.version}</version>
+        </dependency>
+
+
+        <dependency>
             <groupId>com.google.inject</groupId>
             <artifactId>guice</artifactId>
             <scope>provided</scope>

intg/pom.xml

@@ -60,7 +60,6 @@
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <version>${guava.version}</version>
         </dependency>
 
         <dependency>

pom.xml

@@ -459,7 +459,7 @@
         <spring.security.version>3.1.3.RELEASE</spring.security.version>
         <spring-ldap-core.version>1.3.1.RELEASE</spring-ldap-core.version>
         <javax.servlet.version>3.1.0</javax.servlet.version>
-        <guava.version>18.0</guava.version>
+        <guava.version>19.0</guava.version>
 
         <!-- Needed for hooks -->
         <aopalliance.version>1.0</aopalliance.version>
@@ -633,6 +633,12 @@
 
     <dependencyManagement>
         <dependencies>
+
+            <dependency>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>${guava.version}</version>
+            </dependency>
             <!-- AOP dependencies. -->
             <dependency>
                 <groupId>org.aspectj</groupId>

release-log.txt

@@ -9,6 +9,7 @@ ATLAS-1060 Add composite indexes for exact match performance improvements for al
 ATLAS-1127 Modify creation and modification timestamps to Date instead of Long(sumasai)
 
 ALL CHANGES:
+ATLAS-1369 Optimize Gremlin queries generated by DSL translator (jnhagelb)
 ATLAS-1517: updated hive_model to include schema related attributes (sarath.kum4r@gmail.com via mneethiraj)
 ATLAS-1514 Remove duplicates from class array attribute when target is deleted (dkantor)
 ATLAS-1509: fixed issues with deletion during updates (sumasai via mneethiraj)

repository/src/main/java/org/apache/atlas/discovery/DataSetLineageService.java

@@ -18,6 +18,12 @@
 
 package org.apache.atlas.discovery;
 
+import java.util.Arrays;
+import java.util.Iterator;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
 import org.apache.atlas.ApplicationProperties;
 import org.apache.atlas.AtlasClient;
 import org.apache.atlas.AtlasConfiguration;
@@ -25,6 +31,7 @@ import org.apache.atlas.AtlasException;
 import org.apache.atlas.GraphTransaction;
 import org.apache.atlas.discovery.graph.DefaultGraphPersistenceStrategy;
 import org.apache.atlas.discovery.graph.GraphBackedDiscoveryService;
+import org.apache.atlas.query.GremlinQueryResult;
 import org.apache.atlas.query.InputLineageClosureQuery;
 import org.apache.atlas.query.OutputLineageClosureQuery;
 import org.apache.atlas.query.QueryParams;
@@ -42,16 +49,12 @@ import org.apache.atlas.utils.ParamChecker;
 import org.apache.commons.configuration.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import scala.Option;
 import scala.Some;
 import scala.collection.JavaConversions;
 import scala.collection.immutable.List;
 
-import javax.inject.Inject;
-import javax.inject.Singleton;
-import java.util.Arrays;
-import java.util.Iterator;
-
 /**
  * Hive implementation of Lineage service interface.
  */
@@ -139,7 +142,8 @@ public class DataSetLineageService implements LineageService {
                 guid, HIVE_PROCESS_TYPE_NAME,
                 HIVE_PROCESS_INPUT_ATTRIBUTE_NAME, HIVE_PROCESS_OUTPUT_ATTRIBUTE_NAME, Option.empty(),
                 SELECT_ATTRIBUTES, true, graphPersistenceStrategy, graph);
-        return inputsQuery.graph(null).toInstanceJson();
+        GremlinQueryResult result = inputsQuery.evaluate();
+        return inputsQuery.graph(result).toInstanceJson();
     }
 
     @Override
@@ -156,7 +160,8 @@ public class DataSetLineageService implements LineageService {
                 new OutputLineageClosureQuery(AtlasClient.DATA_SET_SUPER_TYPE, SELECT_INSTANCE_GUID, guid, HIVE_PROCESS_TYPE_NAME,
                         HIVE_PROCESS_INPUT_ATTRIBUTE_NAME, HIVE_PROCESS_OUTPUT_ATTRIBUTE_NAME, Option.empty(),
                         SELECT_ATTRIBUTES, true, graphPersistenceStrategy, graph);
-        return outputsQuery.graph(null).toInstanceJson();
+        GremlinQueryResult result = outputsQuery.evaluate();
+        return outputsQuery.graph(result).toInstanceJson();
     }
 
     /**

repository/src/main/java/org/apache/atlas/gremlin/optimizer/AliasFinder.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.FunctionCallExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+import org.apache.atlas.groovy.LiteralExpression;
+import org.apache.atlas.groovy.TraversalStepType;
+
+/**
+ * Finds all aliases in the expression.
+ */
+public class AliasFinder implements CallHierarchyVisitor {
+
+    private List<LiteralExpression> foundAliases = new ArrayList<>();
+
+    //Whether a final alias is needed.  A final alias is needed
+    //if there are transformation steps after the last alias in
+    //the expression.  We initialize this to false since a final
+    //alias is not needed if there are no aliases.
+    private boolean finalAliasNeeded = false;
+
+    @Override
+    public boolean preVisitFunctionCaller(AbstractFunctionExpression expr) {
+        return true;
+    }
+
+    @Override
+    public void visitNonFunctionCaller(GroovyExpression expr) {
+
+    }
+
+    @Override
+    public void visitNullCaller() {
+
+    }
+
+    private static final Set<TraversalStepType> TRANSFORMATION_STEP_TYPES = new HashSet<>(Arrays.asList(
+            TraversalStepType.MAP_TO_ELEMENT,
+            TraversalStepType.MAP_TO_VALUE,
+            TraversalStepType.FLAT_MAP_TO_ELEMENTS,
+            TraversalStepType.FLAT_MAP_TO_VALUES,
+            TraversalStepType.BARRIER,
+            TraversalStepType.NONE));
+
+
+    @Override
+    public boolean postVisitFunctionCaller(AbstractFunctionExpression functionCall) {
+
+        if (functionCall instanceof FunctionCallExpression) {
+            FunctionCallExpression expr = (FunctionCallExpression)functionCall;
+            if (expr.getType() == TraversalStepType.SIDE_EFFECT && expr.getFunctionName().equals("as")) {
+                //We found an alias.  This is currently the last expression we've seen
+                //in our traversal back up the expression tree, so at this point a final
+                //alias is not needed.
+                LiteralExpression aliasNameExpr = (LiteralExpression)expr.getArguments().get(0);
+                foundAliases.add(aliasNameExpr);
+                finalAliasNeeded=false;
+            }
+        }
+
+        if(TRANSFORMATION_STEP_TYPES.contains(functionCall.getType())) {
+            //This step changes the value of the traverser.  Now, a final alias
+            //needs to be added.
+            if(!foundAliases.isEmpty()) {
+                finalAliasNeeded = true;
+            }
+        }
+
+        return true;
+    }
+
+    public List<LiteralExpression> getAliases() {
+        return foundAliases;
+    }
+
+    public boolean isFinalAliasNeeded() {
+
+        return finalAliasNeeded;
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/CallHierarchyVisitor.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+
+/**
+ * Call back interface for visiting the call hierarchy of a function call.
+ */
+public interface CallHierarchyVisitor {
+
+    /**
+     * Visits a function expression before the visit to its caller.
+     *
+     * @param expr
+     *
+     * @return false to terminate the recursion
+     */
+    boolean preVisitFunctionCaller(AbstractFunctionExpression expr);
+
+    /**
+     * Called when a caller that is not an instance of
+     * AbstractFunctionExpression is found. This indicates that the deepest
+     * point in the call hierarchy has been reached.
+     *
+     *
+     */
+    void visitNonFunctionCaller(GroovyExpression expr);
+
+    /**
+     * Called when a null caller is found (this happens for static/user-defined
+     * functions). This indicates that the deepest point in the call hierarchy
+     * has been reached.
+     *
+     */
+    void visitNullCaller();
+
+    /**
+     * Visits a function expression after the visit to its caller.
+     *
+     * @param expr
+     *
+     * @return false to terminate the recursion
+     */
+    boolean postVisitFunctionCaller(AbstractFunctionExpression functionCall);
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/ExpandAndsOptimization.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.atlas.gremlin.GremlinExpressionFactory;
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.FunctionCallExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Optimizer that pulls has expressions out of an 'and' expression.
+ *
+ * For example:
+ *
+ * g.V().and(has('x'),has('y')
+ *
+ * is optimized to:
+ *
+ * g.V().has('x').has('y')
+ *
+ * There are certain cases where it is not safe to move an expression out
+ * of the 'and'.  For example, in the expression
+ *
+ * g.V().and(has('x').out('y'),has('z'))
+ *
+ * has('x').out('y') cannot be moved out of the 'and', since it changes the value of the traverser.
+ *
+ * At this time, the ExpandAndsOptimizer is not able to handle this scenario, so we don't extract
+ * that expression.  In this case, the result is:
+ *
+ * g.V().has('z').and(has('x').out('y'))
+ *
+ * The optimizer will call ExpandAndsOptimization recursively on the children, so
+ * there is no need to recursively update the children here.
+ *
+ * @param expr
+ * @param context
+ * @return the expressions that should be unioned together to get the query result
+ */
+public class ExpandAndsOptimization implements GremlinOptimization {
+
+    private static final Logger logger_ = LoggerFactory.getLogger(ExpandAndsOptimization.class);
+
+
+    private final GremlinExpressionFactory factory;
+
+    public ExpandAndsOptimization(GremlinExpressionFactory factory) {
+        this.factory = factory;
+    }
+
+    @Override
+    public boolean appliesTo(GroovyExpression expr, OptimizationContext contxt) {
+        return expr instanceof FunctionCallExpression && ((FunctionCallExpression)expr).getFunctionName().equals("and");
+    }
+
+    /**
+     * Expands the given and expression.  There is no need to recursively
+     * expand the children here.  This method is called recursively by
+     * GremlinQueryOptimier on the children.
+     *
+     */
+    @Override
+    public GroovyExpression apply(GroovyExpression expr, OptimizationContext context) {
+
+        FunctionCallExpression exprAsFunction = (FunctionCallExpression)expr;
+        GroovyExpression result = exprAsFunction.getCaller();
+
+        List<GroovyExpression> nonExtractableArguments = new ArrayList<>();
+        for(GroovyExpression argument : exprAsFunction.getArguments()) {
+
+            if (GremlinQueryOptimizer.isExtractable(argument)) {
+                //Set the caller of the deepest expression in the call hierarchy
+                //of the argument to point to the current result.
+                //For example, if result is "g.V()" and the updatedArgument is "has('x').has('y')",
+                //updatedArgument would be a tree like this:
+                //
+                //                  has('y')
+                //                 /
+                //                / caller
+                //              |/_
+                //           has('x')
+                //             /
+                //            / caller
+                //          |/_
+                //        (null)
+                //
+                //We would set the caller of has('x') to be g.V(), so result would become g.V().has('x').has('y').
+                //
+                // Note: This operation is currently done by making a copy of the argument tree.  That should
+                // be changed.
+                result = GremlinQueryOptimizer.copyWithNewLeafNode(
+                        (AbstractFunctionExpression) argument, result);
+            } else {
+                logger_.warn("Found non-extractable argument '{}' in the 'and' expression '{}'",argument.toString(), expr.toString());
+                nonExtractableArguments.add(argument);
+            }
+        }
+
+        if (!nonExtractableArguments.isEmpty()) {
+            //add a final 'and' call with the arguments that could not be extracted
+            result =  factory.generateLogicalExpression(result, "and", nonExtractableArguments);
+        }
+        return result;
+    }
+
+    @Override
+    public boolean isApplyRecursively() {
+        return true;
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/ExpressionFinder.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import com.google.common.base.Function;
+
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+
+/**
+ * Call hierarchy visitor that checks if an expression
+ * matching the specified criteria is present
+ * in the call hierarch.
+ */
+public class ExpressionFinder implements CallHierarchyVisitor {
+
+    private final Function<GroovyExpression, Boolean> predicate;
+    private boolean expressionFound = false;
+
+    public ExpressionFinder(Function<GroovyExpression, Boolean> predicate) {
+        this.predicate = predicate;
+    }
+    @Override
+    public boolean preVisitFunctionCaller(AbstractFunctionExpression expr) {
+        if (predicate.apply(expr)) {
+            expressionFound = true;
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public void visitNonFunctionCaller(GroovyExpression expr) {
+        if (predicate.apply(expr)) {
+            expressionFound = true;
+        }
+    }
+
+    @Override
+    public void visitNullCaller() {
+        //nothing to do
+    }
+
+    @Override
+    public boolean postVisitFunctionCaller(AbstractFunctionExpression functionCall) {
+        //nothing to do
+        return true;
+    }
+
+    public boolean isExpressionFound() {
+        return expressionFound;
+    }
+
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/GremlinOptimization.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import org.apache.atlas.groovy.GroovyExpression;
+
+/**
+ * An optimization that can be applied to a gremlin query.
+ */
+public interface GremlinOptimization {
+
+    /**
+     * Whether or not this optimization should be applied to the given expression
+     * @param expr
+     * @param contxt
+     * @return
+     */
+    boolean appliesTo(GroovyExpression expr, OptimizationContext contxt);
+    /**
+     * Whether or not GremlinQueryOptimizer should call this optimization recursively
+     * on the updated children.
+     */
+    boolean isApplyRecursively();
+
+    /**
+     * Applies the optimization.
+     *
+     * @param expr
+     * @param context
+     * @return the optimized expression
+     */
+    GroovyExpression apply(GroovyExpression expr, OptimizationContext context);
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/HasForbiddenType.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import java.util.HashSet;
+import java.util.Set;
+import com.google.common.base.Function;
+
+import org.apache.atlas.gremlin.GremlinExpressionFactory;
+import org.apache.atlas.groovy.GroovyExpression;
+import org.apache.atlas.groovy.TraversalStepType;
+
+/**
+ * Function that tests whether the expression is an 'or'
+ * graph traversal function.
+ */
+public final class HasForbiddenType implements Function<GroovyExpression, Boolean> {
+
+    private Set<TraversalStepType> forbiddenTypes =  new HashSet<>();
+    private final GremlinExpressionFactory factory;
+
+    public HasForbiddenType(GremlinExpressionFactory factory) {
+        this.factory = factory;
+    }
+
+    public void addForbiddenType(TraversalStepType type) {
+        forbiddenTypes.add(type);
+    }
+
+    @Override
+    public Boolean apply(GroovyExpression expr) {
+        if(factory.isLeafAnonymousTraversalExpression(expr)) {
+            return false;
+        }
+        return forbiddenTypes.contains(expr.getType());
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/IsOr.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import com.google.common.base.Function;
+
+import org.apache.atlas.groovy.FunctionCallExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+import org.apache.atlas.groovy.TraversalStepType;
+
+/**
+ * Function that tests whether the expression is an 'or'
+ * graph traversal function.
+ */
+public final class IsOr implements Function<GroovyExpression, Boolean> {
+
+    public static final IsOr INSTANCE = new IsOr();
+
+    private IsOr() {
+    }
+
+    @Override
+    public Boolean apply(GroovyExpression expr) {
+        if (!(expr instanceof FunctionCallExpression)) {
+            return false;
+        }
+        if (expr.getType() != TraversalStepType.FILTER) {
+            return false;
+        }
+        FunctionCallExpression functionCall = (FunctionCallExpression)expr;
+        return functionCall.getFunctionName().equals("or");
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/IsOrParent.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import com.google.common.base.Function;
+
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.FunctionCallExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+import org.apache.atlas.groovy.TraversalStepType;
+
+/**
+ * Matches an expression that gets called after calling or().  For example,
+ * in g.V().or(x,y).toList(), "toList()" is the "or parent", so calling
+ * "apply()" on this expression would return true and calling it on all
+ * the other ones would return false.
+ */
+public final class IsOrParent implements Function<GroovyExpression, Boolean> {
+
+    public static final IsOrParent INSTANCE = new IsOrParent();
+
+    private IsOrParent() {
+
+    }
+
+    @Override
+    public Boolean apply(GroovyExpression expr) {
+        if (!(expr instanceof AbstractFunctionExpression)) {
+            return false;
+        }
+        AbstractFunctionExpression functionCall = (AbstractFunctionExpression)expr;
+        GroovyExpression target = functionCall.getCaller();
+
+        if (!(target instanceof FunctionCallExpression)) {
+            return false;
+        }
+
+        if (target.getType() != TraversalStepType.FILTER) {
+            return false;
+        }
+
+        FunctionCallExpression targetFunction = (FunctionCallExpression)target;
+        return targetFunction.getFunctionName().equals("or");
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/OptimizationContext.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.ClosureExpression;
+import org.apache.atlas.groovy.ClosureExpression.VariableDeclaration;
+import org.apache.atlas.groovy.GroovyExpression;
+import org.apache.atlas.groovy.IdentifierExpression;
+import org.apache.atlas.groovy.ListExpression;
+import org.apache.atlas.groovy.TypeCoersionExpression;
+import org.apache.atlas.groovy.VariableAssignmentExpression;
+
+/**
+ * Maintains state information during gremlin optimization.
+ */
+public class OptimizationContext {
+
+    private static final String TMP_ALIAS_NAME = "__tmp";
+    private static final String FINAL_ALIAS_NAME = "__res";
+    private static final String RESULT_VARIABLE = "r";
+    private final List<GroovyExpression> initialStatements = new ArrayList<>();
+    private GroovyExpression resultExpression = getResultVariable();
+    private int counter = 1;
+    private final Map<String, ClosureExpression> functionBodies = new HashMap<>();
+    private AbstractFunctionExpression rangeExpression;
+
+    public OptimizationContext() {
+
+    }
+
+    /**
+     * @return
+     */
+    public List<GroovyExpression> getInitialStatements() {
+        return initialStatements;
+    }
+
+    public void prependStatement(GroovyExpression expr) {
+        initialStatements.add(0, expr);
+    }
+
+    public String getUniqueFunctionName() {
+        return "f" + (counter++);
+    }
+
+
+    public GroovyExpression getDefineResultVariableStmt() {
+        GroovyExpression castExpression = new TypeCoersionExpression(new ListExpression(), "Set");
+        GroovyExpression resultVarDef =  new VariableAssignmentExpression(RESULT_VARIABLE, castExpression);
+        return resultVarDef;
+
+    }
+    public void setResultExpression(GroovyExpression expr) {
+        resultExpression = expr;
+    }
+
+    public GroovyExpression getResultExpression() {
+        return resultExpression;
+    }
+
+    public GroovyExpression getResultVariable() {
+        return new IdentifierExpression(RESULT_VARIABLE);
+    }
+
+    public ClosureExpression getUserDefinedFunctionBody(String functionName) {
+        return functionBodies.get(functionName);
+    }
+
+    public String addFunctionDefinition(VariableDeclaration decl, GroovyExpression body) {
+        String functionName = getUniqueFunctionName();
+        List<VariableDeclaration> decls = (decl == null) ? Collections.<VariableDeclaration>emptyList() : Collections.singletonList(decl);
+        ClosureExpression bodyClosure = new ClosureExpression(body, decls);
+        VariableAssignmentExpression expr = new VariableAssignmentExpression(functionName, bodyClosure);
+        initialStatements.add(expr);
+        functionBodies.put(functionName, bodyClosure);
+        return functionName;
+    }
+
+    public String getFinalAliasName() {
+        return FINAL_ALIAS_NAME;
+    }
+
+    public String getTempAliasName() {
+        return TMP_ALIAS_NAME;
+    }
+
+    public void setRangeExpression(AbstractFunctionExpression rangeExpression) {
+        this.rangeExpression = rangeExpression;
+    }
+
+    public AbstractFunctionExpression getRangeExpression() {
+        return rangeExpression;
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/OrderFinder.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import org.apache.atlas.gremlin.GremlinExpressionFactory;
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+
+
+/**
+ * Finds order expression in the call hierarchy.
+ *
+ */
+public class OrderFinder implements CallHierarchyVisitor {
+
+    private boolean hasOrderExpression;
+    private GremlinExpressionFactory gremlinFactory;
+
+    public OrderFinder(GremlinExpressionFactory gremlinFactory) {
+        this.gremlinFactory = gremlinFactory;
+    }
+
+    @Override
+    public boolean preVisitFunctionCaller(AbstractFunctionExpression expr) {
+
+        return true;
+    }
+
+    @Override
+    public void visitNonFunctionCaller(GroovyExpression expr) {
+    }
+
+    @Override
+    public void visitNullCaller() {
+    }
+
+    @Override
+    public boolean postVisitFunctionCaller(AbstractFunctionExpression functionCall) {
+
+        if (gremlinFactory.isOrderExpression(functionCall)) {
+            hasOrderExpression = true;
+            return false;
+        }
+        return true;
+    }
+
+
+    public boolean hasOrderExpression() {
+
+        return hasOrderExpression;
+    }
+
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/PathExpressionFinder.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.FunctionCallExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+
+/**
+ * Determines whether an expression contains a path() function.
+ */
+public class PathExpressionFinder implements CallHierarchyVisitor {
+
+    private boolean found = false;
+
+    @Override
+    public boolean preVisitFunctionCaller(AbstractFunctionExpression expr) {
+        if(expr instanceof FunctionCallExpression) {
+            found = ((FunctionCallExpression)expr).getFunctionName().equals("path");
+            if(found) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public void visitNonFunctionCaller(GroovyExpression expr) {
+
+    }
+
+    @Override
+    public void visitNullCaller() {
+
+    }
+
+    public boolean isPathExpressionFound() {
+        return found;
+    }
+
+    @Override
+    public boolean postVisitFunctionCaller(AbstractFunctionExpression functionCall) {
+
+        return false;
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/RangeFinder.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.atlas.gremlin.GremlinExpressionFactory;
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+
+
+/**
+ * Finds all range expressions in the call hierarchy.
+ *
+ */
+public class RangeFinder implements CallHierarchyVisitor {
+
+    private List<AbstractFunctionExpression> rangeExpressions = new ArrayList<>();
+    private GremlinExpressionFactory factory;
+
+    public RangeFinder(GremlinExpressionFactory factory) {
+        this.factory = factory;
+    }
+
+    @Override
+    public boolean preVisitFunctionCaller(AbstractFunctionExpression expr) {
+
+        return true;
+    }
+
+    @Override
+    public void visitNonFunctionCaller(GroovyExpression expr) {
+    }
+
+    @Override
+    public void visitNullCaller() {
+    }
+
+    @Override
+    public boolean postVisitFunctionCaller(AbstractFunctionExpression functionCall) {
+
+        if (factory.isRangeExpression(functionCall)) {
+            rangeExpressions.add(functionCall);
+        }
+        return true;
+    }
+
+    public List<AbstractFunctionExpression> getRangeExpressions() {
+        return rangeExpressions;
+    }
+
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/SplitPointFinder.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.atlas.gremlin.GremlinExpressionFactory;
+import org.apache.atlas.groovy.AbstractFunctionExpression;
+import org.apache.atlas.groovy.FunctionCallExpression;
+import org.apache.atlas.groovy.GroovyExpression;
+import org.apache.atlas.groovy.TraversalStepType;
+
+
+/**
+ * This class finds the first place in the expression where the value of the
+ * traverser is changed from being a vertex to being something else.  This is
+ * important in the "or" optimization logic, since the union operation must be
+ * done on *vertices* in order to preserve the semantics of the query.  In addition,
+ * expressions that have side effects must be moved as well, so that those
+ * side effects will be available to the steps that need them.
+ */
+public class SplitPointFinder implements CallHierarchyVisitor {
+
+    //Any steps that change the traverser value to something that is not a vertex or edge
+    //must be included here, so that the union created by ExpandOrsOptimization
+    //is done over vertices/edges.
+    private static final Set<TraversalStepType> TYPES_REQUIRED_IN_RESULT_EXPRESSION = new HashSet<>(
+            Arrays.asList(
+                    TraversalStepType.BARRIER,
+                    TraversalStepType.BRANCH,
+                    TraversalStepType.SIDE_EFFECT,
+                    TraversalStepType.MAP_TO_VALUE,
+                    TraversalStepType.FLAT_MAP_TO_VALUES,
+                    TraversalStepType.END,
+                    TraversalStepType.NONE));
+
+    private final Set<String> requiredAliases = new HashSet<>();
+
+    //Exceptions to the requirement that all expressions with a type
+    //in the above list must be in the result expression.  If the
+    //function name is in this list, it is ok for that expression
+    //to not be in the result expression.  This mechanism allows
+    //aliases to remain outside the result expression.  Other
+    //exceptions may be found in the future.
+    private static final Map<TraversalStepType, WhiteList> WHITE_LISTS = new HashMap<>();
+    static {
+        WHITE_LISTS.put(TraversalStepType.SIDE_EFFECT, new WhiteList("as"));
+    }
+
+    private final GremlinExpressionFactory factory;
+
+    public SplitPointFinder(GremlinExpressionFactory factory) {
+        this.factory = factory;
+    }
+
+    /**
+     * Represents a set of function names.
+     */
+    private static final class WhiteList {
+        private Set<String> allowedFunctionNames = new HashSet<>();
+        public WhiteList(String... names) {
+            for(String name : names) {
+                allowedFunctionNames.add(name);
+            }
+        }
+        public boolean contains(String name) {
+            return allowedFunctionNames.contains(name);
+        }
+    }
+
+    private AbstractFunctionExpression splitPoint;
+
+    @Override
+    public boolean preVisitFunctionCaller(AbstractFunctionExpression expr) {
+        requiredAliases.addAll(factory.getAliasesRequiredByExpression(expr));
+        return true;
+    }
+
+    @Override
+    public void visitNonFunctionCaller(GroovyExpression expr) {
+
+    }
+
+    @Override
+    public void visitNullCaller() {
+
+    }
+
+    public AbstractFunctionExpression getSplitPoint() {
+        return splitPoint;
+    }
+
+    @Override
+    public boolean postVisitFunctionCaller(AbstractFunctionExpression functionCall) {
+        String aliasName = factory.getAliasNameIfRelevant(functionCall);
+        if (splitPoint == null) {
+
+            boolean required = isRequiredAlias(aliasName) ||
+                    isRequiredInResultExpression(functionCall);
+            if (required) {
+                splitPoint = functionCall;
+            }
+        }
+        removeSeenAlias(aliasName);
+
+        return true;
+    }
+
+    private void removeSeenAlias(String aliasName) {
+        if(aliasName != null) {
+            requiredAliases.remove(aliasName);
+        }
+    }
+
+    private boolean isRequiredAlias(String aliasName) {
+        if(aliasName != null) {
+            return requiredAliases.contains(aliasName);
+        }
+        return false;
+    }
+
+    private boolean isRequiredInResultExpression(AbstractFunctionExpression expr) {
+
+        TraversalStepType type = expr.getType();
+        if (!TYPES_REQUIRED_IN_RESULT_EXPRESSION.contains(type)) {
+            return false;
+        }
+
+        if(expr instanceof FunctionCallExpression) {
+            FunctionCallExpression functionCall = (FunctionCallExpression)expr;
+            //check if the white list permits this function call.  If there is
+            //no white list, all expressions with the current step type must go in the
+            //result expression.
+            WhiteList whiteList = WHITE_LISTS.get(type);
+            if(whiteList != null && whiteList.contains(functionCall.getFunctionName())) {
+                return false;
+            }
+        }
+        return true;
+
+    }
+}

repository/src/main/java/org/apache/atlas/gremlin/optimizer/UpdatedExpressions.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.gremlin.optimizer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.atlas.groovy.GroovyExpression;
+
+/**
+ * Represents a list of updated expressions.
+ */
+public class UpdatedExpressions {
+
+    private List<List<GroovyExpression>> updatedChildren = new ArrayList<>();
+    private boolean changed = false;
+
+    public UpdatedExpressions(boolean changed, List<List<GroovyExpression>> updatedChildren) {
+        this.changed = changed;
+        this.updatedChildren = updatedChildren;
+    }
+
+    public List<List<GroovyExpression>> getUpdatedChildren() {
+        return updatedChildren;
+    }
+
+    public boolean hasChanges() {
+        return changed;
+    }
+}

repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepository.java

@@ -67,19 +67,26 @@ public class GraphBackedMetadataRepository implements MetadataRepository {
 
     private static final GraphHelper graphHelper = GraphHelper.getInstance();
 
-    private final AtlasGraph graph;
-
     private DeleteHandler deleteHandler;
 
-    private GraphToTypedInstanceMapper graphToInstanceMapper;
+    private final IAtlasGraphProvider graphProvider;
+    private final GraphToTypedInstanceMapper graphToInstanceMapper;
 
     @Inject
     public GraphBackedMetadataRepository(DeleteHandler deleteHandler) {
-        this.graph = AtlasGraphProvider.getGraphInstance();
-        graphToInstanceMapper = new GraphToTypedInstanceMapper(graph);
+        this.graphProvider = new AtlasGraphProvider();
+        this.graphToInstanceMapper = new GraphToTypedInstanceMapper(graphProvider);
         this.deleteHandler = deleteHandler;
     }
 
+    //for testing only
+    public GraphBackedMetadataRepository(IAtlasGraphProvider graphProvider, DeleteHandler deleteHandler) {
+        this.graphProvider = graphProvider;
+        this.graphToInstanceMapper = new GraphToTypedInstanceMapper(graphProvider);
+        this.deleteHandler = deleteHandler;
+    }
+
+
     public GraphToTypedInstanceMapper getGraphToInstanceMapper() {
         return graphToInstanceMapper;
     }
@@ -194,7 +201,7 @@ public class GraphBackedMetadataRepository implements MetadataRepository {
             LOG.debug("Retrieving entity list for type={}", entityType);
         }
 
-        AtlasGraphQuery query = graph.query().has(Constants.ENTITY_TYPE_PROPERTY_KEY, entityType);
+        AtlasGraphQuery query = getGraph().query().has(Constants.ENTITY_TYPE_PROPERTY_KEY, entityType);
         Iterator<AtlasVertex> results = query.vertices().iterator();
         if (!results.hasNext()) {
             return Collections.emptyList();
@@ -429,7 +436,7 @@ public class GraphBackedMetadataRepository implements MetadataRepository {
                 requestContext.getDeletedEntityIds());
     }
 
-    public AtlasGraph getGraph() {
-        return AtlasGraphProvider.getGraphInstance();
+    public AtlasGraph getGraph() throws RepositoryException {
+        return graphProvider.get();
     }
 }

repository/src/main/java/org/apache/atlas/repository/graph/GraphToTypedInstanceMapper.java

@@ -19,6 +19,7 @@ package org.apache.atlas.repository.graph;
 
 import com.google.inject.Singleton;
 import org.apache.atlas.AtlasException;
+import org.apache.atlas.repository.RepositoryException;
 import org.apache.atlas.repository.Constants;
 import org.apache.atlas.repository.graphdb.AtlasEdge;
 import org.apache.atlas.repository.graphdb.AtlasEdgeDirection;
@@ -59,10 +60,10 @@ public final class GraphToTypedInstanceMapper {
     private static TypeSystem typeSystem = TypeSystem.getInstance();
     private static final GraphHelper graphHelper = GraphHelper.getInstance();
 
-    private AtlasGraph graph;
+    private final IAtlasGraphProvider graphProvider;
 
-    public GraphToTypedInstanceMapper(AtlasGraph graph) {
-        this.graph = graph;
+    public GraphToTypedInstanceMapper(IAtlasGraphProvider graphProvider) {
+        this.graphProvider = graphProvider;
     }
 
     public ITypedReferenceableInstance mapGraphToTypedInstance(String guid, AtlasVertex instanceVertex)
@@ -407,7 +408,7 @@ public final class GraphToTypedInstanceMapper {
 
 
     public ITypedInstance getReferredEntity(String edgeId, IDataType<?> referredType) throws AtlasException {
-        final AtlasEdge edge = graph.getEdge(edgeId);
+        final AtlasEdge edge = getGraph().getEdge(edgeId);
         if (edge != null) {
             final AtlasVertex referredVertex = edge.getInVertex();
             if (referredVertex != null) {
@@ -433,5 +434,9 @@ public final class GraphToTypedInstanceMapper {
         }
         return null;
     }
+
+    private AtlasGraph getGraph() throws RepositoryException {
+        return graphProvider.get();
+    }
 }
 

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityGraphDiscoveryV1.java

@@ -17,8 +17,14 @@
  */
 package org.apache.atlas.repository.store.graph.v1;
 
-import atlas.shaded.hbase.guava.common.annotations.VisibleForTesting;
-import com.google.inject.Provider;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
 import org.apache.atlas.AtlasErrorCode;
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.TypeCategory;
@@ -26,8 +32,8 @@ import org.apache.atlas.model.instance.AtlasEntity;
 import org.apache.atlas.model.instance.AtlasObjectId;
 import org.apache.atlas.model.instance.AtlasStruct;
 import org.apache.atlas.model.typedef.AtlasStructDef.AtlasAttributeDef;
-import org.apache.atlas.repository.store.graph.EntityGraphDiscoveryContext;
 import org.apache.atlas.repository.store.graph.EntityGraphDiscovery;
+import org.apache.atlas.repository.store.graph.EntityGraphDiscoveryContext;
 import org.apache.atlas.repository.store.graph.EntityResolver;
 import org.apache.atlas.type.AtlasArrayType;
 import org.apache.atlas.type.AtlasEntityType;
@@ -36,14 +42,9 @@ import org.apache.atlas.type.AtlasStructType;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 
-import javax.inject.Inject;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
 
 
 public class AtlasEntityGraphDiscoveryV1 implements EntityGraphDiscovery {

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityStoreV1.java

@@ -18,8 +18,9 @@
 package org.apache.atlas.repository.store.graph.v1;
 
 
-import atlas.shaded.hbase.guava.common.annotations.VisibleForTesting;
-import com.google.inject.Inject;
+import java.util.ArrayList;
+import java.util.List;
+
 import org.apache.atlas.AtlasErrorCode;
 import org.apache.atlas.GraphTransaction;
 import org.apache.atlas.RequestContextV1;
@@ -34,15 +35,17 @@ import org.apache.atlas.model.instance.EntityMutationResponse;
 import org.apache.atlas.model.instance.EntityMutations;
 import org.apache.atlas.repository.graphdb.AtlasVertex;
 import org.apache.atlas.repository.store.graph.AtlasEntityStore;
-import org.apache.atlas.repository.store.graph.EntityGraphDiscoveryContext;
 import org.apache.atlas.repository.store.graph.EntityGraphDiscovery;
+import org.apache.atlas.repository.store.graph.EntityGraphDiscoveryContext;
 import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.List;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.Inject;
+
+
 
 public class AtlasEntityStoreV1 implements AtlasEntityStore {
 

repository/src/main/java/org/apache/atlas/util/AtlasRepositoryConfiguration.java

@@ -24,9 +24,11 @@ import org.apache.atlas.ApplicationProperties;
 import org.apache.atlas.AtlasException;
 import org.apache.atlas.repository.audit.EntityAuditRepository;
 import org.apache.atlas.repository.audit.HBaseBasedAuditRepository;
+import org.apache.atlas.repository.graph.AtlasGraphProvider;
 import org.apache.atlas.repository.graph.DeleteHandler;
 import org.apache.atlas.repository.graph.SoftDeleteHandler;
 import org.apache.atlas.repository.graphdb.GraphDatabase;
+import org.apache.atlas.repository.graphdb.GremlinVersion;
 import org.apache.atlas.repository.store.graph.v1.DeleteHandlerV1;
 import org.apache.atlas.repository.store.graph.v1.SoftDeleteHandlerV1;
 import org.apache.atlas.typesystem.types.cache.DefaultTypeCache;
@@ -137,7 +139,6 @@ public class AtlasRepositoryConfiguration {
         }
     }
 
-
     private static final String GRAPH_DATABASE_IMPLEMENTATION_PROPERTY = "atlas.graphdb.backend";
     private static final String DEFAULT_GRAPH_DATABASE_IMPLEMENTATION_CLASS = "org.apache.atlas.repository.graphdb.titan0.Titan0GraphDatabase";
 
@@ -153,6 +154,22 @@ public class AtlasRepositoryConfiguration {
     }
 
     /**
+     * This optimization is configurable as a fail-safe in case issues are found
+     * with the optimizer in production systems.
+     */
+    public static final String GREMLIN_OPTIMIZER_ENABLED_PROPERTY = "atlas.query.gremlinOptimizerEnabled";
+    private static final boolean DEFAULT_GREMLIN_OPTIMZER_ENABLED = true;
+
+    public static boolean isGremlinOptimizerEnabled() {
+        try {
+            return ApplicationProperties.get().getBoolean(GREMLIN_OPTIMIZER_ENABLED_PROPERTY, DEFAULT_GREMLIN_OPTIMZER_ENABLED);
+        } catch (AtlasException e) {
+            LOG.error("Could not determine value of " + GREMLIN_OPTIMIZER_ENABLED_PROPERTY + ".  Defaulting to " + DEFAULT_GREMLIN_OPTIMZER_ENABLED, e);
+            return DEFAULT_GREMLIN_OPTIMZER_ENABLED;
+        }
+    }
+
+    /**
      * Get the list of operations which are configured to be skipped from auditing
      * Valid format is HttpMethod:URL eg: GET:Version
      * @return list of string

repository/src/main/scala/org/apache/atlas/query/GremlinQuery.scala

@@ -32,15 +32,19 @@ import scala.collection.JavaConversions.bufferAsJavaList
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
+
 import org.apache.atlas.gremlin.GremlinExpressionFactory
+import org.apache.atlas.gremlin.optimizer.GremlinQueryOptimizer
 import org.apache.atlas.groovy.CastExpression
-import org.apache.atlas.groovy.CodeBlockExpression
+import org.apache.atlas.groovy.ClosureExpression
+import org.apache.atlas.groovy.LabeledExpression
 import org.apache.atlas.groovy.FunctionCallExpression
 import org.apache.atlas.groovy.GroovyExpression
 import org.apache.atlas.groovy.GroovyGenerationContext
 import org.apache.atlas.groovy.IdentifierExpression
 import org.apache.atlas.groovy.ListExpression
 import org.apache.atlas.groovy.LiteralExpression
+import org.apache.atlas.groovy.TraversalStepType
 import org.apache.atlas.query.Expressions.AliasExpression
 import org.apache.atlas.query.Expressions.ArithmeticExpression
 import org.apache.atlas.query.Expressions.BackReference
@@ -78,7 +82,10 @@ import org.apache.atlas.query.Expressions.MaxExpression
 import org.apache.atlas.query.Expressions.MinExpression
 import org.apache.atlas.query.Expressions.SumExpression
 import org.apache.atlas.query.Expressions.CountExpression
+
+import org.apache.atlas.util.AtlasRepositoryConfiguration
 import java.util.HashSet
+
 trait IntSequence {
     def next: Int
 }
@@ -120,6 +127,69 @@ trait SelectExpressionHandling {
         }
     }
 
+    // Removes back references in comparison expressions that are
+    // right after an alias expression.
+    //
+    //For example:
+    // .as('x').and(select('x').has(y),...) is changed to
+    // .as('x').and(has(y),...)
+    //
+    //This allows the "has" to be extracted out of the and/or by
+    //the GremlinQueryOptimizer so the index can be used to evaluate
+    //the predicate.
+
+     val RemoveUnneededBackReferences : PartialFunction[Expression, Expression] = {
+
+         case filterExpr@FilterExpression(aliasExpr@AliasExpression(_,aliasName), filterChild) => {
+             val updatedChild = removeUnneededBackReferences(filterChild, aliasName)
+             val changed  = !(updatedChild eq filterChild)
+             if(changed) {
+                 FilterExpression(aliasExpr, updatedChild)
+             }
+             else {
+                 filterExpr
+             }
+
+         }
+         case x => x
+     }
+     def removeUnneededBackReferences(expr: Expression, outerAlias: String) : Expression = expr match {
+         case logicalExpr@LogicalExpression(logicalOp,children) => {
+            var changed : Boolean = false;
+            val updatedChildren : List[Expression] = children.map { child =>
+                val updatedChild = removeUnneededBackReferences(child, outerAlias);
+                changed |= ! (updatedChild eq child);
+                updatedChild
+            }
+           if(changed) {
+               LogicalExpression(logicalOp,updatedChildren)
+           }
+           else {
+              logicalExpr
+           }
+        }
+        case comparisonExpr@ComparisonExpression(_,_,_) => {
+            var changed = false
+             val updatedLeft = removeUnneededBackReferences(comparisonExpr.left, outerAlias);
+             changed |= !( updatedLeft eq comparisonExpr.left);
+
+             val updatedRight = removeUnneededBackReferences(comparisonExpr.right, outerAlias);
+             changed |= !(updatedRight eq comparisonExpr.right);
+
+             if (changed) {
+                 ComparisonExpression(comparisonExpr.symbol, updatedLeft, updatedRight)
+             } else {
+                 comparisonExpr
+             }
+        }
+        case FieldExpression(fieldName, fieldInfo, Some(br @ BackReference(brAlias, _, _))) if outerAlias.equals(brAlias) => {
+           //Remove the back reference, since the thing it references is right in front
+           //of the comparison expression we're in
+           FieldExpression(fieldName, fieldInfo, None)
+       }
+       case x => x
+    }
+
     //in groupby, convert alias expressions defined in the group by child to BackReferences
     //in the groupby list and selectList.
     val AddBackReferencesToGroupBy : PartialFunction[Expression, Expression] = {
@@ -456,7 +526,10 @@ class GremlinTranslator(expr: Expression,
             return translateLiteralValue(l.dataType, l);
         }
         case list: ListLiteral[_] =>  {
-            val  values : java.util.List[GroovyExpression] = translateList(list.rawValue, true); //why hard coded
+            //Here, we are creating a Groovy list literal expression ([value1, value2, value3]).  Because
+            //of this, any gremlin query expressions within the list must start with an anonymous traversal.
+            //We set 'inClosure' to true in this case to make that happen.
+            val  values : java.util.List[GroovyExpression] = translateList(list.rawValue, true);
             return new ListExpression(values);
         }
         case in@TraitInstanceExpression(child) => {
@@ -493,7 +566,7 @@ class GremlinTranslator(expr: Expression,
         case limitOffset@LimitExpression(child, limit, offset) => {
             val childExpr = genQuery(parent, child, inClosure);
             val totalResultRows = limit.value + offset.value;
-            return GremlinExpressionFactory.INSTANCE.generateLimitExpression(childExpr, offset.value, totalResultRows);
+            return GremlinExpressionFactory.INSTANCE.generateRangeExpression(childExpr, offset.value, totalResultRows);
         }
         case count@CountExpression() => {
           val listExpr = GremlinExpressionFactory.INSTANCE.getClosureArgumentValue();
@@ -621,8 +694,7 @@ class GremlinTranslator(expr: Expression,
 
     def genFullQuery(expr: Expression, hasSelect: Boolean): String = {
 
-        var q : GroovyExpression = new FunctionCallExpression(new IdentifierExpression("g"),"V");
-
+        var q : GroovyExpression = new FunctionCallExpression(TraversalStepType.START, new IdentifierExpression(TraversalStepType.SOURCE, "g"),"V");
 
         val debug:Boolean = false
         if(gPersistenceBehavior.addGraphVertexPrefix(preStatements)) {
@@ -631,15 +703,23 @@ class GremlinTranslator(expr: Expression,
 
         q = genQuery(q, expr, false)
 
-        q = new FunctionCallExpression(q, "toList");
+        q = GremlinExpressionFactory.INSTANCE.generateToListExpression(q);
         q = gPersistenceBehavior.getGraph().addOutputTransformationPredicate(q, hasSelect, expr.isInstanceOf[PathExpression]);
 
-        var overallExpression = new CodeBlockExpression();
-        overallExpression.addStatements(preStatements);
-        overallExpression.addStatement(q)
-        overallExpression.addStatements(postStatements);
 
-        var qryStr = generateGremlin(overallExpression);
+        if(AtlasRepositoryConfiguration.isGremlinOptimizerEnabled()) {
+            q = GremlinQueryOptimizer.getInstance().optimize(q);
+        }
+
+	    val closureExpression = new ClosureExpression();
+
+        closureExpression.addStatements(preStatements);
+        closureExpression.addStatement(q)
+        closureExpression.addStatements(postStatements);
+
+	    val overallExpression = new LabeledExpression("L", closureExpression);
+
+        val qryStr = generateGremlin(overallExpression);
 
         if(debug) {
           println(" query " + qryStr)
@@ -666,6 +746,7 @@ class GremlinTranslator(expr: Expression,
         e1 = e1.transformUp(addAliasToLoopInput())
         e1 = e1.transformUp(instanceClauseToTop(e1))
         e1 = e1.transformUp(traitClauseWithInstanceForTop(e1))
+        e1 = e1.transformUp(RemoveUnneededBackReferences)
 
         //Following code extracts the select expressions from expression tree.
 

repository/src/test/java/org/apache/atlas/discovery/GraphBackedDiscoveryServiceTest.java

@@ -508,6 +508,9 @@ public class GraphBackedDiscoveryServiceTest extends BaseRepositoryTest {
                 {"from hive_db limit 3 offset 1", 2},
                 {"hive_db", 3},
                 {"hive_db where hive_db.name=\"Reporting\"", 1},
+                {"hive_db where hive_db.name=\"Reporting\" or hive_db.name=\"Sales\" or hive_db.name=\"Logging\" limit 1 offset 1", 1},
+                {"hive_db where hive_db.name=\"Reporting\" or hive_db.name=\"Sales\" or hive_db.name=\"Logging\" limit 1 offset 2", 1},
+                {"hive_db where hive_db.name=\"Reporting\" or hive_db.name=\"Sales\" or hive_db.name=\"Logging\" limit 2 offset 1", 2},
                 {"hive_db where hive_db.name=\"Reporting\" limit 10 ", 1},
                 {"hive_db hive_db.name = \"Reporting\"", 1},
                 {"hive_db where hive_db.name=\"Reporting\" select name, owner", 1},

repository/src/test/java/org/apache/atlas/repository/graph/TestIntSequence.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.atlas.repository.graph;
+
+import org.apache.atlas.query.IntSequence;
+
+/**
+ * IntSequence for use in unit tests.
+ *
+ */
+public class TestIntSequence implements IntSequence {
+
+    public static final IntSequence INSTANCE = new TestIntSequence();
+    private TestIntSequence() {
+    }
+    @Override
+    public int next() {
+        return 0;
+    }
+}
\ No newline at end of file

repository/src/test/scala/org/apache/atlas/query/GremlinTest.scala

@@ -886,7 +886,7 @@ class GremlinTest extends BaseGremlinTest {
         .or(id("name").`=`(string("Reporting")))).field("Table").as("tab")
         .select(id("db1").field("name").as("dbName"), id("tab").field("name").as("tabName")), g, gp
     )
-    validateJson(r, "{\n  \"query\":\"DB as db1 where (db1.createTime > 0) or (name = \\\"Reporting\\\") Table as tab select db1.name as dbName, tab.name as tabName\",\n  \"dataType\":{\n    \"typeName\":\"__tempQueryResultStruct6\",\n    \"attributeDefinitions\":[\n      {\n        \"name\":\"dbName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      },\n      {\n        \"name\":\"tabName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      }\n    ]\n  },\n  \"rows\":[\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"sales_fact\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"product_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"time_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"customer_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_daily_mv\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_monthly_mv\"\n    }\n  ]\n}")
+    validateJson(r, "{\n  \"query\":\"DB as db1 where (createTime > 0) or (name = \\\"Reporting\\\") Table as tab select db1.name as dbName, tab.name as tabName\",\n  \"dataType\":{\n    \"typeName\":\"__tempQueryResultStruct6\",\n    \"attributeDefinitions\":[\n      {\n        \"name\":\"dbName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      },\n      {\n        \"name\":\"tabName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      }\n    ]\n  },\n  \"rows\":[\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"sales_fact\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"product_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"time_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"customer_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_daily_mv\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct6\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_monthly_mv\"\n    }\n  ]\n}")
   }
 
   @Test def testJoinAndSelect3 {
@@ -896,7 +896,7 @@ class GremlinTest extends BaseGremlinTest {
         .or(id("db1").hasField("owner"))).field("Table").as("tab")
         .select(id("db1").field("name").as("dbName"), id("tab").field("name").as("tabName")), g, gp
     )
-    validateJson(r, "{\n  \"query\":\"DB as db1 where (db1.createTime > 0) and (db1.name = \\\"Reporting\\\") or db1 has owner Table as tab select db1.name as dbName, tab.name as tabName\",\n  \"dataType\":{\n    \"typeName\":\"__tempQueryResultStruct7\",\n    \"attributeDefinitions\":[\n      {\n        \"name\":\"dbName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      },\n      {\n        \"name\":\"tabName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      }\n    ]\n  },\n  \"rows\":[\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"sales_fact\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"product_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"time_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"customer_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_daily_mv\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_monthly_mv\"\n    }\n  ]\n}")
+    validateJson(r, "{\n  \"query\":\"DB as db1 where (createTime > 0) and (name = \\\"Reporting\\\") or db1 has owner Table as tab select db1.name as dbName, tab.name as tabName\",\n  \"dataType\":{\n    \"typeName\":\"__tempQueryResultStruct7\",\n    \"attributeDefinitions\":[\n      {\n        \"name\":\"dbName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      },\n      {\n        \"name\":\"tabName\",\n        \"dataTypeName\":\"string\",\n        \"multiplicity\":{\n          \"lower\":0,\n          \"upper\":1,\n          \"isUnique\":false\n        },\n        \"isComposite\":false,\n        \"isUnique\":false,\n        \"isIndexable\":false,\n        \"reverseAttributeName\":null\n      }\n    ]\n  },\n  \"rows\":[\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"sales_fact\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"product_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"time_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Sales\",\n      \"tabName\":\"customer_dim\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_daily_mv\"\n    },\n    {\n      \"$typeName$\":\"__tempQueryResultStruct7\",\n      \"dbName\":\"Reporting\",\n      \"tabName\":\"sales_fact_monthly_mv\"\n    }\n  ]\n}")
   }
 
   @Test def testJoinAndSelect4 {

shaded/hbase-client-shaded/pom.xml

@@ -37,6 +37,15 @@
         </dependency>
     </dependencies>
 
+    <dependencyManagement>
+      <dependencies>
+	<dependency>
+          <groupId>com.google.guava</groupId>
+          <artifactId>guava</artifactId>
+          <version>12.0.1</version>
+	</dependency>
+      </dependencies>
+    </dependencyManagement>
     <build>
         <plugins>
             <plugin>

shaded/hbase-server-shaded/pom.xml

@@ -42,6 +42,15 @@
             <artifactId>hbase-server</artifactId>
         </dependency>
     </dependencies>
+    <dependencyManagement>
+      <dependencies>
+	<dependency>
+          <groupId>com.google.guava</groupId>
+          <artifactId>guava</artifactId>
+          <version>12.0.1</version>
+	</dependency>
+      </dependencies>
+    </dependencyManagement>
     
     <build>
         <plugins>