Gh 3065 java pom fix

pom.xml

@@ -48,6 +48,8 @@
         <append.shaded.classifier />
 
         <java.version>1.8</java.version>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
 
         <!-- Dependency version properties -->
         <junit.version>5.9.3</junit.version>

store-implementation/federated-store/src/main/java/uk/gov/gchq/gaffer/federatedstore/FederatedStore.java

@@ -54,7 +54,7 @@
 import uk.gov.gchq.gaffer.federatedstore.operation.handler.impl.FederatedRemoveGraphAndDeleteAllDataHandler;
 import uk.gov.gchq.gaffer.federatedstore.operation.handler.impl.FederatedRemoveGraphHandler;
 import uk.gov.gchq.gaffer.federatedstore.schema.FederatedViewValidator;
-import uk.gov.gchq.gaffer.federatedstore.util.ApplyViewToElementsFunction;
+import uk.gov.gchq.gaffer.federatedstore.util.FederatedElementFunction;
 import uk.gov.gchq.gaffer.federatedstore.util.MergeSchema;
 import uk.gov.gchq.gaffer.graph.GraphSerialisable;
 import uk.gov.gchq.gaffer.operation.Operation;
@@ -156,8 +156,8 @@ public FederatedStore(@JsonProperty("customPropertiesAuths") final Set<String> c
         this.storeConfiguredMergeFunctions = (null == storeConfiguredMergeFunctions) ? new HashMap<>() : new HashMap<>(storeConfiguredMergeFunctions);
 
         this.storeConfiguredMergeFunctions.putIfAbsent(GetTraits.class.getCanonicalName(), new CollectionIntersect<>());
-        this.storeConfiguredMergeFunctions.putIfAbsent(GetAllElements.class.getCanonicalName(), new ApplyViewToElementsFunction());
-        this.storeConfiguredMergeFunctions.putIfAbsent(GetElements.class.getCanonicalName(), new ApplyViewToElementsFunction());
+        this.storeConfiguredMergeFunctions.putIfAbsent(GetAllElements.class.getCanonicalName(), new FederatedElementFunction());
+        this.storeConfiguredMergeFunctions.putIfAbsent(GetElements.class.getCanonicalName(), new FederatedElementFunction());
         this.storeConfiguredMergeFunctions.putIfAbsent(GetSchema.class.getCanonicalName(), new MergeSchema());
     }
 

store-implementation/federated-store/src/main/java/uk/gov/gchq/gaffer/federatedstore/util/ApplyViewToElementsFunction.java → store-implementation/federated-store/src/main/java/uk/gov/gchq/gaffer/federatedstore/util/FederatedElementFunction.java

@@ -25,28 +25,42 @@
 import uk.gov.gchq.gaffer.core.exception.GafferRuntimeException;
 import uk.gov.gchq.gaffer.data.element.Element;
 import uk.gov.gchq.gaffer.data.elementdefinition.view.View;
+import uk.gov.gchq.gaffer.data.elementdefinition.view.ViewElementDefinition;
 import uk.gov.gchq.gaffer.graph.Graph;
 import uk.gov.gchq.gaffer.graph.GraphConfig;
+import uk.gov.gchq.gaffer.graph.GraphSerialisable;
+import uk.gov.gchq.gaffer.mapstore.MapStore;
 import uk.gov.gchq.gaffer.mapstore.MapStoreProperties;
 import uk.gov.gchq.gaffer.operation.OperationException;
 import uk.gov.gchq.gaffer.operation.impl.add.AddElements;
 import uk.gov.gchq.gaffer.operation.impl.get.GetAllElements;
 import uk.gov.gchq.gaffer.store.Context;
 import uk.gov.gchq.gaffer.store.schema.Schema;
+import uk.gov.gchq.gaffer.store.schema.SchemaElementDefinition;
 import uk.gov.gchq.gaffer.user.User;
+import uk.gov.gchq.koryphe.tuple.predicate.TupleAdaptedPredicate;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
+import java.util.stream.Stream;
 
-public class ApplyViewToElementsFunction implements ContextSpecificMergeFunction<Object, Iterable<Object>, Iterable<Object>> {
-    private static final Logger LOGGER = LoggerFactory.getLogger(ApplyViewToElementsFunction.class);
+/**
+ * This class is used to address some of the issues with having the elements distributes amongst multiple graphs.
+ * Such as the re-application of View filter or Schema Validation after the local aggregation of results from multiple graphs.
+ * By default, a local in memory MapStore is used for local aggregation,
+ * but a Graph or {@link GraphSerialisable} of any kind could be supplied via the {@link #context} with the key {@link #TEMP_RESULTS_GRAPH}.
+ */
+public class FederatedElementFunction implements ContextSpecificMergeFunction<Object, Iterable<Object>, Iterable<Object>> {
+    private static final Logger LOGGER = LoggerFactory.getLogger(FederatedElementFunction.class);
     public static final String VIEW = "view";
     public static final String SCHEMA = "schema";
     public static final String USER = "user";
@@ -56,16 +70,16 @@ public class ApplyViewToElementsFunction implements ContextSpecificMergeFunction
     @JsonProperty("context")
     private Map<String, Object> context;
 
-    public ApplyViewToElementsFunction() {
+    public FederatedElementFunction() {
     }
 
-    public ApplyViewToElementsFunction(final Map<String, Object> context) throws GafferCheckedException {
+    public FederatedElementFunction(final Map<String, Object> context) throws GafferCheckedException {
         this();
         try {
             // Check if results graph, hasn't already be supplied, otherwise make a default results graph.
             if (!context.containsKey(TEMP_RESULTS_GRAPH)) {
                 final Graph resultsGraph = new Graph.Builder()
-                        .config(new GraphConfig(String.format("%s%s%d", TEMP_RESULTS_GRAPH, ApplyViewToElementsFunction.class.getSimpleName(), RANDOM.nextInt(Integer.MAX_VALUE))))
+                        .config(new GraphConfig(String.format("%s%s%d", TEMP_RESULTS_GRAPH, FederatedElementFunction.class.getSimpleName(), RANDOM.nextInt(Integer.MAX_VALUE))))
                         .addSchema((Schema) context.get(SCHEMA))
                         //MapStore easy in memory Store. Large results size may not be suitable, a graph could be provided via Context.
                         .addStoreProperties(new MapStoreProperties())
@@ -77,16 +91,37 @@ public ApplyViewToElementsFunction(final Map<String, Object> context) throws Gaf
             }
             // Validate the supplied context before using
             validate(context);
+
+            updateViewWithValidationFromSchema(context);
+
             this.context = Collections.unmodifiableMap(context);
         } catch (final Exception e) {
             throw new GafferCheckedException("Unable to create TemporaryResultsGraph", e);
         }
 
     }
 
+    private static void updateViewWithValidationFromSchema(final Map<String, Object> context) {
+        //Only do this for MapStore, not required for other stores.
+        if (MapStore.class.getName().equals(getGraph(context).getStoreProperties().getStoreClass())) {
+            //Update View with
+            final View view = (View) context.get(VIEW);
+            final Schema schema = (Schema) context.get(SCHEMA);
+            final View.Builder updatedView = new View.Builder(view);
+
+            //getUpdatedDefs and add to new view.
+            getUpdatedViewDefsFromSchemaDefs(schema.getEdges(), view)
+                    .forEach(e -> updatedView.edge(e.getKey(), e.getValue()));
+            getUpdatedViewDefsFromSchemaDefs(schema.getEntities(), view)
+                    .forEach(e -> updatedView.entity(e.getKey(), e.getValue()));
+
+            context.put(VIEW, updatedView.build());
+        }
+    }
+
     @Override
-    public ApplyViewToElementsFunction createFunctionWithContext(final HashMap<String, Object> context) throws GafferCheckedException {
-        return new ApplyViewToElementsFunction(context);
+    public FederatedElementFunction createFunctionWithContext(final HashMap<String, Object> context) throws GafferCheckedException {
+        return new FederatedElementFunction(context);
     }
 
     /**
@@ -111,7 +146,8 @@ private static void validate(final Map<String, Object> context) {
 
         if (!context.containsKey(TEMP_RESULTS_GRAPH)) {
             throw new IllegalStateException("Error: context invalid, did not contain a Temporary Results Graph.");
-        } else if (!(context.get(TEMP_RESULTS_GRAPH) instanceof Graph)) {
+        } else if (!(context.get(TEMP_RESULTS_GRAPH) instanceof Graph)
+                && !(context.get(TEMP_RESULTS_GRAPH) instanceof GraphSerialisable)) {
             throw new IllegalArgumentException(String.format("Error: context invalid, value for %s was not a Graph, found: %s", TEMP_RESULTS_GRAPH, context.get(TEMP_RESULTS_GRAPH)));
         }
 
@@ -120,6 +156,38 @@ private static void validate(final Map<String, Object> context) {
         }
     }
 
+    private static Stream<Map.Entry<String, ViewElementDefinition>> getUpdatedViewDefsFromSchemaDefs(final Map<String, ? extends SchemaElementDefinition> groupDefs, final View view) {
+        return groupDefs.entrySet().stream()
+                .map(e -> new AbstractMap.SimpleImmutableEntry<>(e.getKey(), getUpdatedViewDefFromSchemaDef(e.getKey(), e.getValue(), view)));
+    }
+
+    private static ViewElementDefinition getUpdatedViewDefFromSchemaDef(final String groupName, final SchemaElementDefinition schemaElementDef, final View view) {
+        final ViewElementDefinition.Builder updatePreAggregationFilter;
+        final ArrayList<TupleAdaptedPredicate<String, ?>> updatedFilterFunctions = new ArrayList<>();
+
+        //Add Schema Validation
+        if (schemaElementDef.hasValidation()) {
+            updatedFilterFunctions.addAll(schemaElementDef.getValidator().getComponents());
+        }
+
+        if (view != null) {
+            final ViewElementDefinition viewElementDef = view.getElement(groupName);
+            //Add View Validation
+            if (viewElementDef != null && viewElementDef.hasPostAggregationFilters()) {
+                updatedFilterFunctions.addAll(viewElementDef.getPostAggregationFilter().getComponents());
+            }
+            //Init Builder with contents of the view.
+            updatePreAggregationFilter = new ViewElementDefinition.Builder(viewElementDef);
+        } else {
+            updatePreAggregationFilter = new ViewElementDefinition.Builder();
+        }
+
+        //override
+        updatePreAggregationFilter.postAggregationFilterFunctions(updatedFilterFunctions);
+
+        return updatePreAggregationFilter.build();
+    }
+
     @Override
     @JsonIgnore
     public Set<String> getRequiredContextValues() {
@@ -137,7 +205,7 @@ public Iterable<Object> apply(final Object update, final Iterable<Object> state)
             }
         }
 
-        final Graph resultsGraph = (Graph) context.get(TEMP_RESULTS_GRAPH);
+        final Graph resultsGraph = getGraph(context);
         final Context userContext = new Context((User) context.get(USER));
         try {
             // The update object might be a lazy AccumuloElementRetriever and might be MASSIVE.
@@ -152,4 +220,12 @@ public Iterable<Object> apply(final Object update, final Iterable<Object> state)
             throw new GafferRuntimeException("Error getting all elements from temporary graph, due to:" + e.getMessage(), e);
         }
     }
+
+    private static Graph getGraph(final Map<String, Object> context) {
+        final Object g = context.get(TEMP_RESULTS_GRAPH);
+        final Graph resultsGraph = g instanceof GraphSerialisable
+                ? ((GraphSerialisable) g).getGraph()
+                : (Graph) g;
+        return resultsGraph;
+    }
 }

store-implementation/federated-store/src/main/java/uk/gov/gchq/gaffer/federatedstore/util/FederatedStoreUtil.java

@@ -66,9 +66,9 @@
 
 import static java.util.Objects.isNull;
 import static java.util.Objects.nonNull;
-import static uk.gov.gchq.gaffer.federatedstore.util.ApplyViewToElementsFunction.SCHEMA;
-import static uk.gov.gchq.gaffer.federatedstore.util.ApplyViewToElementsFunction.USER;
-import static uk.gov.gchq.gaffer.federatedstore.util.ApplyViewToElementsFunction.VIEW;
+import static uk.gov.gchq.gaffer.federatedstore.util.FederatedElementFunction.SCHEMA;
+import static uk.gov.gchq.gaffer.federatedstore.util.FederatedElementFunction.USER;
+import static uk.gov.gchq.gaffer.federatedstore.util.FederatedElementFunction.VIEW;
 
 public final class FederatedStoreUtil {
     private static final Logger LOGGER = LoggerFactory.getLogger(FederatedStoreUtil.class);

store-implementation/federated-store/src/test/java/uk/gov/gchq/gaffer/federatedstore/FederatedElementFunctionAggregationSchemaValidationTest.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2023 Crown Copyright
+ *
+ * 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 uk.gov.gchq.gaffer.federatedstore;
+
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import uk.gov.gchq.gaffer.data.element.Entity;
+import uk.gov.gchq.gaffer.federatedstore.operation.FederatedOperation;
+import uk.gov.gchq.gaffer.operation.OperationException;
+import uk.gov.gchq.gaffer.operation.impl.add.AddElements;
+import uk.gov.gchq.gaffer.operation.impl.get.GetAllElements;
+import uk.gov.gchq.gaffer.store.schema.Schema;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.BASIC_VERTEX;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.GRAPH_ID_A;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.GRAPH_ID_B;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.GRAPH_ID_C;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.GRAPH_ID_TEST_FEDERATED_STORE;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.GROUP_BASIC_ENTITY;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.PROPERTY_1;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.addGraphToAccumuloStore;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.contextTestUser;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.loadSchemaFromJson;
+import static uk.gov.gchq.gaffer.federatedstore.FederatedStoreTestUtil.resetForFederatedTests;
+
+public class FederatedElementFunctionAggregationSchemaValidationTest {
+
+    private FederatedStore federatedStore;
+    private Entity entity1, entity99, entityOther;
+
+    @BeforeEach
+    public void before() throws Exception {
+        resetForFederatedTests();
+
+        federatedStore = new FederatedStore();
+        federatedStore.initialise(GRAPH_ID_TEST_FEDERATED_STORE, new Schema(), new FederatedStoreProperties());
+
+        entity1 = new Entity.Builder()
+                .group(GROUP_BASIC_ENTITY)
+                .vertex(BASIC_VERTEX)
+                .property(PROPERTY_1, 1)
+                .build();
+
+        entity99 = new Entity.Builder()
+                .group(GROUP_BASIC_ENTITY)
+                .vertex(BASIC_VERTEX)
+                .property(PROPERTY_1, 99)
+                .build();
+
+        entityOther = new Entity.Builder()
+                .group(GROUP_BASIC_ENTITY)
+                .vertex("basicVertexOther")
+                .property(PROPERTY_1, 99)
+                .build();
+    }
+
+    @Test
+    public void shouldOnlyReturn1EntitySmallerThanSchemaValidationLimit() throws Exception {
+
+        //given
+        addGraphToAccumuloStore(federatedStore, GRAPH_ID_A, true, loadSchemaFromJson("/schema/basicEntityValidateLess100Schema.json"));
+        addGraphToAccumuloStore(federatedStore, GRAPH_ID_B, true, loadSchemaFromJson("/schema/basicEntityValidateLess100Schema.json"));
+
+        addEntity(GRAPH_ID_A, entity1);
+        addEntity(GRAPH_ID_B, entity99);
+        addEntity(GRAPH_ID_B, entityOther);
+
+        //when
+        final Iterable elementsWithPropertyLessThan100 = federatedStore.execute(new GetAllElements(), contextTestUser());
+
+        //then
+        assertThat(elementsWithPropertyLessThan100)
+                .isNotNull()
+                .withFailMessage("should not return entity \"basicVertex\" with un-aggregated property 1 or 99")
+                .doesNotContain(entity1, entity99)
+                .withFailMessage("should not return entity \"basicVertex\" with an aggregated property 100, which is less than view filter 100")
+                .doesNotContain(new Entity.Builder()
+                        .group(GROUP_BASIC_ENTITY)
+                        .vertex(BASIC_VERTEX)
+                        .property(PROPERTY_1, 100)
+                        .build())
+                .withFailMessage("should return entity \"basicVertexOther\" with property 99, which is less than view filter 100")
+                .containsExactly(entityOther);
+    }
+
+    @Test
+    public void shouldNotReturnAnyElementsAfterInValidationInTemporaryMap() throws Exception {
+
+        //given
+        addGraphToAccumuloStore(federatedStore, GRAPH_ID_A, true, loadSchemaFromJson("/schema/basicEntityValidateLess100Schema.json"));
+        addGraphToAccumuloStore(federatedStore, GRAPH_ID_B, true, loadSchemaFromJson("/schema/basicEntityValidateLess100Schema.json"));
+        addGraphToAccumuloStore(federatedStore, GRAPH_ID_C, true, loadSchemaFromJson("/schema/basicEntityValidateLess100Schema.json"));
+
+        addEntity(GRAPH_ID_A, entity99); // 99 is valid
+        addEntity(GRAPH_ID_B, entity1); // 100 is not valid.
+        addEntity(GRAPH_ID_C, entity1); // correct behavior 100 & 1 is invalid. returning 1 would be incorrect if 100 had been deleted.
+        addEntity(GRAPH_ID_B, entityOther);
+
+        //when
+        final Iterable elementsWithPropertyLessThan100 = federatedStore.execute(new GetAllElements(), contextTestUser());
+
+        //then
+        assertThat(elementsWithPropertyLessThan100)
+                .isNotNull()
+                .withFailMessage("should not return entity \"basicVertex\" with un-aggregated property 1 or 99")
+                .doesNotContain(entity1, entity99)
+                .withFailMessage("should not return entity \"basicVertex\" with an aggregated property 100, which is less than view filter 100")
+                .doesNotContain(new Entity.Builder()
+                        .group(GROUP_BASIC_ENTITY)
+                        .vertex(BASIC_VERTEX)
+                        .property(PROPERTY_1, 100)
+                        .build())
+                .withFailMessage("should return entity \"basicVertexOther\" with property 99, which is less than view filter 100")
+                .containsExactly(entityOther)
+                .hasSize(1);
+    }
+
+
+    private void addEntity(final String graphIdA, final Entity entity) throws OperationException {
+        federatedStore.execute(new FederatedOperation.Builder()
+                .op(new AddElements.Builder()
+                        .input(entity)
+                        .build())
+                .graphIdsCSV(graphIdA)
+                .build(), contextTestUser());
+    }
+
+}