[DE-1071] Filter pushdown (#20)

This PR implements filter pushdown optimization for ArangoDB TinkerPop integration by translating Gremlin predicates into AQL queries. The implementation adds query optimization to reduce data transfer and improve performance.

This PR implements the push down only for filters of type: org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasStep.

For example this optimizes queries like g.V().has("value", "foo"), by reading from the database only the matching vertices.

Author: rashtao

spotbugs/spotbugs-exclude.xml

@@ -35,4 +35,9 @@
         <Bug pattern="SE_BAD_FIELD"/>
     </Match>
 
+    <Match>
+        <Class name="com.arangodb.tinkerpop.gremlin.process.traversal.strategy.optimization.ArangoStepStrategy"/>
+        <Bug pattern="SING_SINGLETON_IMPLEMENTS_SERIALIZABLE"/>
+    </Match>
+
 </FindBugsFilter>

src/main/java/com/arangodb/tinkerpop/gremlin/client/ArangoDBGraphClient.java

@@ -18,6 +18,7 @@
 
 import java.util.*;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import com.arangodb.*;
 import com.arangodb.entity.*;
@@ -28,12 +29,14 @@
 import com.arangodb.serde.jackson.JacksonSerde;
 import com.arangodb.tinkerpop.gremlin.persistence.*;
 import com.arangodb.tinkerpop.gremlin.persistence.serde.SerdeModule;
+import com.arangodb.tinkerpop.gremlin.process.filter.ArangoFilter;
+import com.arangodb.tinkerpop.gremlin.process.filter.EmptyFilter;
+import com.arangodb.tinkerpop.gremlin.process.filter.FilterSupport;
 import com.arangodb.tinkerpop.gremlin.structure.*;
 import com.arangodb.tinkerpop.gremlin.utils.AqlDeserializer;
 import com.fasterxml.jackson.databind.*;
 import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
 import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -110,15 +113,25 @@ public void updateGraphVariables(VariablesData document) {
         }
     }
 
+    public Stream<VertexData> getGraphVertices(List<ElementId> ids, ArangoFilter filter, Set<String> colNames) {
+        logger.debug("Get all {} graph vertices, filtered by AQL filters", config.graphName);
+        return getGraphDocuments(ids, filter, colNames, VertexData.class);
+    }
+
+    public Stream<EdgeData> getGraphEdges(List<ElementId> ids, ArangoFilter filter, Set<String> colNames) {
+        logger.debug("Get all {} graph edges, filtered by AQL filters", config.graphName);
+        return getGraphDocuments(ids, filter, colNames, EdgeData.class);
+    }
+
     /**
      * Get vertices of a graph. If no ids are provided, get all vertices.
      *
      * @param ids the ids to match
      * @return the documents
      */
-    public ArangoIterable<VertexData> getGraphVertices(final List<ElementId> ids) {
+    public Stream<VertexData> getGraphVertices(final List<ElementId> ids) {
         logger.debug("Get all {} graph vertices, filtered by ids: {}", config.graphName, ids);
-        return getGraphDocuments(ids, config.vertices, VertexData.class);
+        return getGraphDocuments(ids, EmptyFilter.instance(), config.vertices, VertexData.class);
     }
 
     /**
@@ -127,19 +140,28 @@ public ArangoIterable<VertexData> getGraphVertices(final List<ElementId> ids) {
      * @param ids the ids to match
      * @return the documents
      */
-    public ArangoIterable<EdgeData> getGraphEdges(List<ElementId> ids) {
+    public Stream<EdgeData> getGraphEdges(List<ElementId> ids) {
         logger.debug("Get all {} graph edges, filtered by ids: {}", config.graphName, ids);
-        return getGraphDocuments(ids, config.edges, EdgeData.class);
+        return getGraphDocuments(ids, EmptyFilter.instance(), config.edges, EdgeData.class);
     }
 
-    private <V> ArangoIterable<V> getGraphDocuments(List<ElementId> ids, Set<String> colNames, Class<V> clazz) {
+    private <V> Stream<V> getGraphDocuments(List<ElementId> ids, ArangoFilter filter, Set<String> colNames, Class<V> clazz) {
         if (ids.isEmpty()) {
-            return query(ArangoDBQueryBuilder.readAllDocuments(colNames), clazz, null);
+            if (colNames.isEmpty()) {
+                return Stream.empty();
+            }
+            return query(ArangoDBQueryBuilder.readAllDocuments(colNames, filter), clazz, null);
         } else {
             List<ElementId> prunedIds = ids.stream()
                     .filter(it -> colNames.contains(it.getCollection()))
                     .collect(Collectors.toList());
-            return query("FOR d IN DOCUMENT(@ids) RETURN d", clazz, Collections.singletonMap("ids", prunedIds));
+            StringBuilder query = new StringBuilder();
+            query.append("FOR d IN DOCUMENT(@ids)");
+            if (filter.getSupport() != FilterSupport.NONE) {
+                query.append(" FILTER ").append(filter.toAql("d"));
+            }
+            query.append(" RETURN d");
+            return query(query.toString(), clazz, Collections.singletonMap("ids", prunedIds));
         }
     }
 
@@ -191,19 +213,19 @@ public ArangoDB getArangoDriver() {
         return db.arango();
     }
 
-    public Iterator<Object> query(final String query, final Map<String, ?> parameters, final AqlQueryOptions options) {
-        Iterator<JsonNode> res = query(query, JsonNode.class, parameters, options);
-        return IteratorUtils.map(res, aqlDeserializer::deserialize);
+    public Stream<Object> query(final String query, final Map<String, ?> parameters, final AqlQueryOptions options) {
+        return query(query, JsonNode.class, parameters, options)
+                .map(aqlDeserializer::deserialize);
     }
 
-    private <V> ArangoCursor<V> query(String query, Class<V> type, Map<String, ?> parameters) {
+    private <V> Stream<V> query(String query, Class<V> type, Map<String, ?> parameters) {
         return query(query, type, parameters, new AqlQueryOptions());
     }
 
-    private <V> ArangoCursor<V> query(String query, Class<V> type, Map<String, ?> parameters, AqlQueryOptions options) {
+    private <V> Stream<V> query(String query, Class<V> type, Map<String, ?> parameters, AqlQueryOptions options) {
         logger.debug("Executing AQL query: {}, with parameters: {}, with options: {}", query, parameters, options);
         try {
-            return db.query(query, type, parameters, options);
+            return db.query(query, type, parameters, options).stream();
         } catch (ArangoDBException e) {
             throw mapException(e);
         }
@@ -303,7 +325,7 @@ public void updateVertex(ArangoDBVertex vertex) {
         logger.debug("Document updated, new rev {}", vertexEntity.getRev());
     }
 
-    public Iterator<VertexData> getVertexNeighbors(ElementId vertexId, Set<String> edgeCollections, Direction direction, String[] labels) {
+    public Stream<VertexData> getVertexNeighbors(ElementId vertexId, Set<String> edgeCollections, Direction direction, String[] labels) {
         logger.debug("Get vertex {}:{} Neighbors, in {}, from collections {}", vertexId, direction, config.graphName, edgeCollections);
         String query = ArangoDBQueryBuilder.readVertexNeighbors(config.graphName, direction, config, labels);
         Map<String, Object> params = new HashMap<>();
@@ -315,7 +337,7 @@ public Iterator<VertexData> getVertexNeighbors(ElementId vertexId, Set<String> e
         return query(query, VertexData.class, params);
     }
 
-    public Iterator<EdgeData> getVertexEdges(ElementId vertexId, Set<String> edgeCollections, Direction direction, String[] labels) {
+    public Stream<EdgeData> getVertexEdges(ElementId vertexId, Set<String> edgeCollections, Direction direction, String[] labels) {
         logger.debug("Get vertex {}:{} Edges, in {}, from collections {}", vertexId, direction, config.graphName, edgeCollections);
         String query = ArangoDBQueryBuilder.readVertexEdges(config.graphName, direction, config, labels);
         Map<String, Object> params = new HashMap<>();

src/main/java/com/arangodb/tinkerpop/gremlin/client/ArangoDBQueryBuilder.java

@@ -19,6 +19,8 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import com.arangodb.tinkerpop.gremlin.process.filter.ArangoFilter;
+import com.arangodb.tinkerpop.gremlin.process.filter.FilterSupport;
 import com.arangodb.tinkerpop.gremlin.structure.ArangoDBGraphConfig;
 import org.apache.tinkerpop.gremlin.structure.Direction;
 
@@ -57,26 +59,33 @@ private static StringBuilder oneStepTraversal(String graphName, Direction direct
         return query;
     }
 
-    public static String readAllDocuments(Set<String> collections) {
+    static String readAllDocuments(Set<String> collections, ArangoFilter filter) {
         if (collections.isEmpty()) {
             throw new IllegalArgumentException();
         } else if (collections.size() == 1) {
-            return readAllDocumentsFromSingleCollection(collections.iterator().next());
+            return readAllDocumentsFromSingleCollection(collections.iterator().next(), filter);
         } else {
-            return readAllDocumentsFromMultipleCollections(collections);
+            return readAllDocumentsFromMultipleCollections(collections, filter);
         }
     }
 
-    private static String readAllDocumentsFromMultipleCollections(Set<String> collections) {
+    private static String readAllDocumentsFromMultipleCollections(Set<String> collections, ArangoFilter filter) {
         String inner = collections.stream()
-                .map(ArangoDBQueryBuilder::readAllDocumentsFromSingleCollection)
+                .map(it -> ArangoDBQueryBuilder.readAllDocumentsFromSingleCollection(it, filter))
                 .map(it -> "(" + it + ")")
                 .collect(Collectors.joining(","));
         return String.format("FOR d in UNION(%s) RETURN d", inner);
     }
 
-    private static String readAllDocumentsFromSingleCollection(String collection) {
-        return String.format("FOR x IN %s RETURN x", escape(collection));
+    private static String readAllDocumentsFromSingleCollection(String collection, ArangoFilter filter) {
+        StringBuilder query = new StringBuilder()
+                .append("FOR x IN ")
+                .append(escape(collection));
+        if (filter.getSupport() != FilterSupport.NONE) {
+            query.append(" FILTER ").append(filter.toAql("x"));
+        }
+        query.append(" RETURN x");
+        return query.toString();
     }
 
     private static String escape(String collection) {

src/main/java/com/arangodb/tinkerpop/gremlin/process/filter/AndFilter.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2025 ArangoDB GmbH and The University of York
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.arangodb.tinkerpop.gremlin.process.filter;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class AndFilter implements ArangoFilter {
+    private final List<ArangoFilter> filters;
+
+    public static ArangoFilter of(Collection<ArangoFilter> filters) {
+        List<ArangoFilter> supportedFilters = filters.stream()
+                .filter(it -> it.getSupport() != FilterSupport.NONE)
+                .collect(Collectors.toList());
+        if (supportedFilters.isEmpty()) {
+            return EmptyFilter.instance();
+        } else if (supportedFilters.size() == 1) {
+            return supportedFilters.get(0);
+        } else {
+            return new AndFilter(supportedFilters);
+        }
+    }
+
+    private AndFilter(List<ArangoFilter> filters) {
+        this.filters = filters;
+    }
+
+    /**
+     * +---------++---------+---------+---------+
+     * |   AND   ||  FULL   | PARTIAL |  NONE   |
+     * +---------++---------+---------+---------+
+     * | FULL    || FULL    | PARTIAL | PARTIAL |
+     * | PARTIAL || PARTIAL | PARTIAL | PARTIAL |
+     * | NONE    || PARTIAL | PARTIAL | NONE    |
+     * +---------++---------+---------+---------+
+     */
+    @Override
+    public FilterSupport getSupport() {
+        if (filters.stream().map(ArangoFilter::getSupport).allMatch(FilterSupport.FULL::equals)) {
+            return FilterSupport.FULL;
+        } else {
+            return FilterSupport.PARTIAL;
+        }
+    }
+
+    @Override
+    public String toAql(String variableName) {
+        return filters.stream()
+                .map(it -> it.toAql(variableName))
+                .collect(Collectors.joining(" AND ", "(", ")"));
+    }
+}

src/main/java/com/arangodb/tinkerpop/gremlin/process/filter/ArangoFilter.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2025 ArangoDB GmbH and The University of York
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.arangodb.tinkerpop.gremlin.process.filter;
+
+import org.apache.tinkerpop.gremlin.process.traversal.P;
+import org.apache.tinkerpop.gremlin.process.traversal.util.AndP;
+import org.apache.tinkerpop.gremlin.process.traversal.util.OrP;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+public interface ArangoFilter {
+    static ArangoFilter of(String key, P<?> p) {
+        String pName = p.getPredicateName();
+        switch (pName) {
+            case "eq":
+                return new CompareEqFilter(key, p.getValue());
+            case "neq":
+                return NotFilter.of(new CompareEqFilter(key, p.getValue()));
+            case "lt":
+                return new CompareLtFilter(key, p.getValue());
+            case "lte":
+                return OrFilter.of(Arrays.asList(new CompareLtFilter(key, p.getValue()), new CompareEqFilter(key, p.getValue())));
+            case "gt":
+                return new CompareGtFilter(key, p.getValue());
+            case "gte":
+                return OrFilter.of(Arrays.asList(new CompareGtFilter(key, p.getValue()), new CompareEqFilter(key, p.getValue())));
+            case "within":
+                return new ContainsWithinFilter(key, (Collection<?>) p.getValue());
+            case "without":
+                return NotFilter.of(new ContainsWithinFilter(key, (Collection<?>) p.getValue()));
+            case "containing":
+                return new TextContainingFilter(key, (String) p.getValue());
+            case "notContaining":
+                return NotFilter.of(new TextContainingFilter(key, (String) p.getValue()));
+            case "endingWith":
+                return new TextRegexFilter(key, Pattern.quote((String) p.getValue()) + "$");
+            case "notEndingWith":
+                return NotFilter.of(new TextRegexFilter(key, Pattern.quote((String) p.getValue()) + "$"));
+            case "startingWith":
+                return new TextStartingWithFilter(key, (String) p.getValue());
+            case "notStartingWith":
+                return NotFilter.of(new TextStartingWithFilter(key, (String) p.getValue()));
+            case "regex":
+                return new TextRegexFilter(key, (String) p.getValue());
+            case "notRegex":
+                return NotFilter.of(new TextRegexFilter(key, (String) p.getValue()));
+            case "or":
+                if (p instanceof OrP) {
+                    return OrFilter.of(((OrP<?>) p).getPredicates().stream()
+                            .map(it -> of(key, it))
+                            .collect(Collectors.toList()));
+                }
+                throw new UnsupportedOperationException("Unsupported predicate: " + p);
+            case "and":
+                if (p instanceof AndP) {
+                    return AndFilter.of(((AndP<?>) p).getPredicates().stream()
+                            .map(it -> of(key, it))
+                            .collect(Collectors.toList()));
+                }
+                throw new UnsupportedOperationException("Unsupported predicate: " + p);
+            default:
+                throw new UnsupportedOperationException("Unsupported predicate: " + p);
+        }
+    }
+
+    FilterSupport getSupport();
+
+    String toAql(String variableName);
+}

src/main/java/com/arangodb/tinkerpop/gremlin/process/filter/CompareEqFilter.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2025 ArangoDB GmbH and The University of York
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.arangodb.tinkerpop.gremlin.process.filter;
+
+
+import com.arangodb.tinkerpop.gremlin.process.value.Value;
+
+import java.util.Objects;
+
+public class CompareEqFilter implements ArangoFilter {
+
+    private final String attribute;
+    private final Value value;
+
+    public CompareEqFilter(String attribute, Object value) {
+        Objects.requireNonNull(attribute, "attribute cannot be null");
+        if (attribute.isEmpty()) {
+            throw new IllegalArgumentException("attribute cannot be empty");
+        }
+        this.attribute = attribute;
+        this.value = Value.of(value);
+    }
+
+    @Override
+    public FilterSupport getSupport() {
+        return value.getSupport();
+    }
+
+    @Override
+    public String toAql(String variableName) {
+        return "`" + variableName + "`.`" + attribute + "` == " + value.toAql();
+    }
+}