Begin with implementation of Executors

Author: tobias-weber

plugins/workflow-engine/src/main/java/org/polypheny/db/workflow/dag/Workflow.java

@@ -18,7 +18,7 @@
 
 import java.util.List;
 import java.util.UUID;
-import org.polypheny.db.workflow.dag.activities.Activity;
+import org.polypheny.db.workflow.dag.activities.ActivityWrapper;
 import org.polypheny.db.workflow.dag.edges.Edge;
 import org.polypheny.db.workflow.models.EdgeModel;
 import org.polypheny.db.workflow.models.WorkflowConfigModel;
@@ -29,7 +29,9 @@
  */
 public interface Workflow {
 
-    List<Activity> getActivities(); // TODO: change return type to Map<UUID, Activity> ?
+    List<ActivityWrapper> getActivities(); // TODO: change return type to Map<UUID, Activity> ?
+
+    ActivityWrapper getActivity( UUID activityId );
 
     /**
      * Get all edges of this workflow as list with arbitrary order.
@@ -47,7 +49,7 @@ public interface Workflow {
      */
     List<Edge> getEdges( UUID from, UUID to );
 
-    List<Edge> getEdges( Activity from, Activity to );
+    List<Edge> getEdges( ActivityWrapper from, ActivityWrapper to );
 
     List<Edge> getInEdges( UUID target );
 
@@ -69,7 +71,7 @@ public interface Workflow {
      */
     void setState( WorkflowState state );
 
-    void addActivity( Activity activity );
+    void addActivity( ActivityWrapper activity );
 
     void deleteActivity( UUID activityId );
 

plugins/workflow-engine/src/main/java/org/polypheny/db/workflow/dag/WorkflowImpl.java

@@ -26,7 +26,7 @@
 import lombok.Setter;
 import org.polypheny.db.catalog.exceptions.GenericRuntimeException;
 import org.polypheny.db.util.Pair;
-import org.polypheny.db.workflow.dag.activities.Activity;
+import org.polypheny.db.workflow.dag.activities.ActivityWrapper;
 import org.polypheny.db.workflow.dag.edges.Edge;
 import org.polypheny.db.workflow.models.ActivityModel;
 import org.polypheny.db.workflow.models.EdgeModel;
@@ -35,7 +35,7 @@
 
 public class WorkflowImpl implements Workflow {
 
-    private final Map<UUID, Activity> activities;
+    private final Map<UUID, ActivityWrapper> activities;
     private final Map<Pair<UUID, UUID>, List<Edge>> edges;
     private final WorkflowConfigModel config;
     @Getter
@@ -50,7 +50,7 @@ public WorkflowImpl() {
     }
 
 
-    private WorkflowImpl( Map<UUID, Activity> activities, Map<Pair<UUID, UUID>, List<Edge>> edges, WorkflowConfigModel config ) {
+    private WorkflowImpl( Map<UUID, ActivityWrapper> activities, Map<Pair<UUID, UUID>, List<Edge>> edges, WorkflowConfigModel config ) {
         this.activities = activities;
         this.edges = edges;
         this.config = config;
@@ -59,11 +59,11 @@ private WorkflowImpl( Map<UUID, Activity> activities, Map<Pair<UUID, UUID>, List
 
     public static Workflow fromModel( WorkflowModel model ) {
 
-        Map<UUID, Activity> activities = new ConcurrentHashMap<>();
+        Map<UUID, ActivityWrapper> activities = new ConcurrentHashMap<>();
         Map<Pair<UUID, UUID>, List<Edge>> edges = new ConcurrentHashMap<>();
 
         for ( ActivityModel a : model.getActivities() ) {
-            activities.put( a.getId(), Activity.fromModel( a ) );
+            activities.put( a.getId(), ActivityWrapper.fromModel( a ) );
         }
         for ( EdgeModel e : model.getEdges() ) {
             Pair<UUID, UUID> key = Pair.of( e.getFromId(), e.getToId() );
@@ -76,11 +76,17 @@ public static Workflow fromModel( WorkflowModel model ) {
 
 
     @Override
-    public List<Activity> getActivities() {
+    public List<ActivityWrapper> getActivities() {
         return new ArrayList<>( activities.values() );
     }
 
 
+    @Override
+    public ActivityWrapper getActivity( UUID activityId ) {
+        return activities.get( activityId );
+    }
+
+
     @Override
     public List<Edge> getEdges() {
         return edges.values()
@@ -97,7 +103,7 @@ public List<Edge> getEdges( UUID from, UUID to ) {
 
 
     @Override
-    public List<Edge> getEdges( Activity from, Activity to ) {
+    public List<Edge> getEdges( ActivityWrapper from, ActivityWrapper to ) {
         return getEdges( from.getId(), to.getId() );
     }
 
@@ -133,7 +139,7 @@ public WorkflowConfigModel getConfig() {
 
 
     @Override
-    public void addActivity( Activity activity ) {
+    public void addActivity( ActivityWrapper activity ) {
         if ( activities.containsKey( activity.getId() ) ) {
             throw new GenericRuntimeException( "Cannot add activity instance that is already part of this workflow." );
         }

plugins/workflow-engine/src/main/java/org/polypheny/db/workflow/dag/activities/Activity.java

@@ -16,36 +16,18 @@
 
 package org.polypheny.db.workflow.dag.activities;
 
-import com.fasterxml.jackson.databind.JsonNode;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.UUID;
 import org.polypheny.db.algebra.type.AlgDataType;
+import org.polypheny.db.catalog.logistic.DataModel;
 import org.polypheny.db.workflow.dag.settings.SettingDef.SettingValue;
+import org.polypheny.db.workflow.dag.variables.WritableVariableStore;
 import org.polypheny.db.workflow.engine.execution.ExecutionContext;
 import org.polypheny.db.workflow.engine.storage.CheckpointReader;
-import org.polypheny.db.workflow.models.ActivityConfigModel;
-import org.polypheny.db.workflow.models.ActivityModel;
-import org.polypheny.db.workflow.models.RenderModel;
 
 public interface Activity {
 
-    String getType();
-
-    UUID getId();
-
-    ActivityState getState();
-
-    /**
-     * Changes the state of this activity to the specified state.
-     * After initialization, this should never be done by the activity itself.
-     * The state is typically changed by the scheduler.
-     *
-     * @param state the new state of this activity
-     */
-    void setState( ActivityState state );
-
     /**
      * This method computes the output tuple-types by considering (a preview of) input types and settings.
      * If the input types or settings are not available or cannot be validated yet, the output type is set to an empty {@link Optional}
@@ -61,32 +43,39 @@ public interface Activity {
      */
     List<Optional<AlgDataType>> previewOutTypes( List<Optional<AlgDataType>> inTypes, Map<String, Optional<SettingValue>> settings ) throws ActivityException;
 
-    void execute( List<CheckpointReader> inputs, Map<String, SettingValue> settings, ExecutionContext ctx ) throws Exception; // default execution method. TODO: introduce execution context to track progress, abort, inputs, outputs...
-
-    void updateSettings( Map<String, JsonNode> settings );
-
-    ActivityConfigModel getConfig();
 
-    void setConfig( ActivityConfigModel config );
-
-    RenderModel getRendering();
+    /**
+     * This method is called just before execution starts and can be used to write variables based on input tuple types and settings.
+     * To be able to update variables while having access to the input data, the activity should instead implement {@link VariableWriter}.
+     *
+     * @param inTypes a list of {@link AlgDataType} representing the input tuple types.
+     * @param settings a map of setting keys to {@link SettingValue} representing the settings.
+     * @param variables a WritableVariableStore to be used for updating any variable values.
+     */
+    default void updateVariables( List<AlgDataType> inTypes, Map<String, SettingValue> settings, WritableVariableStore variables ) {
+    }
 
-    void setRendering( RenderModel rendering );
+    // settings do NOT include values from the updateVariables step.
 
-    ActivityDef getDef();
+    /**
+     * Execute this activity.
+     * Any input CheckpointReaders are provided and expected to be closed by the caller.
+     * CheckpointWriters for any outputs are created from the ExecutionContext.
+     * The settings do not incorporate any changes to variables from {@code  updateVariables()}.
+     *
+     * @param inputs a list of input readers for each input specified by the annotation.
+     * @param settings the instantiated setting values, according to the specified settings annotations
+     * @param ctx ExecutionContext to be used for creating checkpoints, updating progress and periodically checking for an abort
+     * @throws Exception in case the execution fails at any point
+     */
+    void execute( List<CheckpointReader> inputs, Map<String, SettingValue> settings, ExecutionContext ctx ) throws Exception; // default execution method
 
     /**
      * Reset any execution-specific state of this activity.
      * It is guaranteed to be called before execution starts.
      */
     void reset();
 
-    ActivityModel toModel( boolean includeState );
-
-    static Activity fromModel( ActivityModel model ) {
-        return ActivityRegistry.activityFromModel( model );
-    }
-
     enum PortType {
         ANY,
         REL,
@@ -102,17 +91,21 @@ public boolean canReadFrom( PortType other ) {
         public boolean canWriteTo( PortType other ) {
             return this == other || other == ANY;
         }
-    }
 
 
-    enum ActivityState {
-        IDLE,
-        QUEUED,
-        EXECUTING,
-        SKIPPED,  // => execution was aborted
-        FAILED,
-        FINISHED,
-        SAVED  // => finished + checkpoint created
+        /**
+         * Returns the corresponding DataModel enum.
+         * In case of ANY, DataModel.RELATIONAL is returned.
+         *
+         * @return the corresponding DataModel
+         */
+        public DataModel getDataModel() {
+            return switch ( this ) {
+                case ANY, REL -> DataModel.RELATIONAL;
+                case DOC -> DataModel.DOCUMENT;
+                case LPG -> DataModel.GRAPH;
+            };
+        }
     }
 
 

plugins/workflow-engine/src/main/java/org/polypheny/db/workflow/dag/activities/ActivityException.java

@@ -21,10 +21,10 @@
 @Getter
 public class ActivityException extends Exception {
 
-    private final Activity activity;
+    private final ActivityWrapper activity;
 
 
-    public ActivityException( String message, Activity activity ) {
+    public ActivityException( String message, ActivityWrapper activity ) {
         super( message );
         this.activity = activity;
     }
@@ -42,7 +42,7 @@ public static class InvalidSettingException extends ActivityException {
         private final String settingKey;
 
 
-        public InvalidSettingException( String message, Activity activity, String settingKey ) {
+        public InvalidSettingException( String message, ActivityWrapper activity, String settingKey ) {
             super( message, activity );
             this.settingKey = settingKey;
         }
@@ -61,7 +61,7 @@ public static class InvalidInputException extends ActivityException {
         private final int index;
 
 
-        public InvalidInputException( String message, Activity activity, int index ) {
+        public InvalidInputException( String message, ActivityWrapper activity, int index ) {
             super( message, activity );
             this.index = index;
         }

plugins/workflow-engine/src/main/java/org/polypheny/db/workflow/dag/activities/ActivityRegistry.java

@@ -37,7 +37,6 @@
 import org.polypheny.db.workflow.dag.annotations.Group;
 import org.polypheny.db.workflow.dag.settings.SettingDef;
 import org.polypheny.db.workflow.dag.settings.SettingDef.SettingValue;
-import org.polypheny.db.workflow.models.ActivityModel;
 import org.reflections.Reflections;
 
 public class ActivityRegistry {
@@ -69,13 +68,13 @@ public static ActivityDef get( String activityType ) {
     }
 
 
-    public static Activity activityFromModel( ActivityModel model ) {
-        ActivityDef def = get( model.getType() );
+    public static Activity activityFromType( String activityType ) {
+        ActivityDef def = get( activityType );
         try {
-            Constructor<? extends Activity> constructor = def.getActivityClass().getConstructor( ActivityModel.class );
-            return constructor.newInstance( model );
+            Constructor<? extends Activity> constructor = def.getActivityClass().getConstructor();
+            return constructor.newInstance();
         } catch ( InvocationTargetException | NoSuchMethodException | InstantiationException | IllegalAccessException e ) {
-            throw new RuntimeException( "Encountered problem during instantiation for type: " + model.getType() );
+            throw new RuntimeException( "Encountered problem during activity instantiation for type: " + activityType );
         }
     }