Support output strategies in pipeline code (#3312)

* Add visitor to output strategies

* Support output strategy in pipeline templates

* Update ts model

* Add license

* Fix checkstyle

streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/compact/PersistPipelineHandler.java

@@ -94,6 +94,7 @@ private List<CompactPipelineElement> makeTemplateConfig(AdapterDescription adapt
         DATA_LAKE_CONNECTOR_ID,
         adapterDescription.getCorrespondingDataStreamElementId(),
         null,
+        null,
         null
     ));
     return pipelineElements;

streampipes-model/src/main/java/org/apache/streampipes/model/output/AppendOutputStrategy.java

@@ -26,15 +26,18 @@
 
 public class AppendOutputStrategy extends OutputStrategy {
 
-  private static final long serialVersionUID = 7202888911899551012L;
-
   private List<EventProperty> eventProperties;
 
   public AppendOutputStrategy() {
     super();
     eventProperties = new ArrayList<>();
   }
 
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
+
   public AppendOutputStrategy(AppendOutputStrategy other) {
     super(other);
     this.setEventProperties(new Cloner().properties(other.getEventProperties()));

streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomOutputStrategy.java

@@ -23,8 +23,6 @@
 
 public class CustomOutputStrategy extends OutputStrategy {
 
-  private static final long serialVersionUID = -5858193127308435472L;
-
   private List<String> selectedPropertyKeys;
 
   private boolean outputRight;
@@ -77,4 +75,9 @@ public List<String> getAvailablePropertyKeys() {
   public void setAvailablePropertyKeys(List<String> availablePropertyKeys) {
     this.availablePropertyKeys = availablePropertyKeys;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }

streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomTransformOutputStrategy.java

@@ -32,7 +32,6 @@ public CustomTransformOutputStrategy() {
     this.eventProperties = new ArrayList<>();
   }
 
-
   public CustomTransformOutputStrategy(CustomTransformOutputStrategy other) {
     super(other);
     this.eventProperties = new Cloner().properties(other.getEventProperties());
@@ -45,4 +44,9 @@ public List<EventProperty> getEventProperties() {
   public void setEventProperties(List<EventProperty> eventProperties) {
     this.eventProperties = eventProperties;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }

streampipes-model/src/main/java/org/apache/streampipes/model/output/FixedOutputStrategy.java

@@ -50,5 +50,9 @@ public void setEventProperties(List<EventProperty> eventProperties) {
     this.eventProperties = eventProperties;
   }
 
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 
 }

streampipes-model/src/main/java/org/apache/streampipes/model/output/KeepOutputStrategy.java

@@ -63,4 +63,9 @@ public boolean isKeepBoth() {
   public void setKeepBoth(boolean keepBoth) {
     this.keepBoth = keepBoth;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }

streampipes-model/src/main/java/org/apache/streampipes/model/output/ListOutputStrategy.java

@@ -46,5 +46,9 @@ public void setPropertyName(String propertyName) {
     this.propertyName = propertyName;
   }
 
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 
 }

streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategy.java

@@ -75,4 +75,6 @@ public List<PropertyRenameRule> getRenameRules() {
   public void setRenameRules(List<PropertyRenameRule> renameRules) {
     this.renameRules = renameRules;
   }
+
+  public abstract void accept(OutputStrategyVisitor visitor);
 }

streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategyVisitor.java

@@ -0,0 +1,38 @@
+/*
+ * 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.streampipes.model.output;
+
+public interface OutputStrategyVisitor {
+
+  void visit(AppendOutputStrategy appendOutputStrategy);
+
+  void visit(CustomOutputStrategy customOutputStrategy);
+
+  void visit(CustomTransformOutputStrategy customTransformOutputStrategy);
+
+  void visit(FixedOutputStrategy fixedOutputStrategy);
+
+  void visit(KeepOutputStrategy keepOutputStrategy);
+
+  void visit(ListOutputStrategy listOutputStrategy);
+
+  void visit(TransformOutputStrategy transformOutputStrategy);
+
+  void visit(UserDefinedOutputStrategy userDefinedOutputStrategy);
+}

streampipes-model/src/main/java/org/apache/streampipes/model/output/TransformOutputStrategy.java

@@ -43,4 +43,9 @@ public List<TransformOperation> getTransformOperations() {
   public void setTransformOperations(List<TransformOperation> transformOperations) {
     this.transformOperations = transformOperations;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }

streampipes-model/src/main/java/org/apache/streampipes/model/output/UserDefinedOutputStrategy.java

@@ -42,4 +42,9 @@ public List<EventProperty> getEventProperties() {
   public void setEventProperties(List<EventProperty> eventProperties) {
     this.eventProperties = eventProperties;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }

streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/CompactPipelineElement.java

@@ -28,5 +28,6 @@ public record CompactPipelineElement(String type,
                                      String ref,
                                      String id,
                                      List<String> connectedTo,
-                                     List<Map<String, Object>> configuration) {
+                                     List<Map<String, Object>> configuration,
+                                     OutputConfiguration output) {
 }

streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/OutputConfiguration.java

@@ -0,0 +1,28 @@
+/*
+ * 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.streampipes.model.pipeline.compact;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+import java.util.List;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public record OutputConfiguration(List<String> keep,
+                                  List<UserDefinedOutput> userDefined) {
+}

streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/UserDefinedOutput.java

@@ -0,0 +1,24 @@
+/*
+ * 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.streampipes.model.pipeline.compact;
+
+public record UserDefinedOutput(String fieldName,
+                                String runtimeType,
+                                String semanticType) {
+}

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/CompactPipelineConverter.java

@@ -19,8 +19,15 @@
 package org.apache.streampipes.manager.pipeline.compact.generation;
 
 import org.apache.streampipes.manager.template.CompactConfigGenerator;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+import org.apache.streampipes.model.output.OutputStrategy;
+import org.apache.streampipes.model.output.UserDefinedOutputStrategy;
 import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.compact.CompactPipelineElement;
+import org.apache.streampipes.model.pipeline.compact.OutputConfiguration;
+import org.apache.streampipes.model.pipeline.compact.UserDefinedOutput;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
 
 import java.util.ArrayList;
@@ -38,6 +45,7 @@ public List<CompactPipelineElement> convert(Pipeline pipeline) {
             stream.getDom(),
             stream.getElementId(),
             null,
+            null,
             null))
         .forEach(pipelineElements::add);
 
@@ -47,7 +55,8 @@ public List<CompactPipelineElement> convert(Pipeline pipeline) {
             processor.getDom(),
             processor.getAppId(),
             processor.getConnectedTo(),
-            getConfig(processor.getStaticProperties())))
+            getConfig(processor.getStaticProperties()),
+            getOutput(processor.getOutputStrategies().get(0))))
         .forEach(pipelineElements::add);
 
     pipeline.getActions().stream()
@@ -56,7 +65,8 @@ public List<CompactPipelineElement> convert(Pipeline pipeline) {
             sink.getDom(),
             sink.getAppId(),
             sink.getConnectedTo(),
-            getConfig(sink.getStaticProperties())))
+            getConfig(sink.getStaticProperties()),
+            null))
         .forEach(pipelineElements::add);
 
     return pipelineElements;
@@ -66,11 +76,12 @@ private CompactPipelineElement createElement(String type,
                                                String ref,
                                                String elementId,
                                                List<String> connectedTo,
-                                               List<Map<String, Object>> config) {
+                                               List<Map<String, Object>> config,
+                                               OutputConfiguration outputConfiguration) {
     var connections = connectedTo != null ? connectedTo.stream()
         .map(this::replaceId)
         .toList() : null;
-    return new CompactPipelineElement(type, replaceId(ref), elementId, connections, config);
+    return new CompactPipelineElement(type, replaceId(ref), elementId, connections, config, outputConfiguration);
   }
 
   public List<Map<String, Object>> getConfig(List<StaticProperty> staticProperties) {
@@ -79,6 +90,26 @@ public List<Map<String, Object>> getConfig(List<StaticProperty> staticProperties
     return configs;
   }
 
+  public OutputConfiguration getOutput(OutputStrategy outputStrategy) {
+    if (outputStrategy instanceof CustomOutputStrategy) {
+      return new OutputConfiguration(((CustomOutputStrategy) outputStrategy).getSelectedPropertyKeys(), null);
+    } else if (outputStrategy instanceof UserDefinedOutputStrategy) {
+      return new OutputConfiguration(
+          null,
+          toCustomConfig(((UserDefinedOutputStrategy) outputStrategy).getEventProperties())
+      );
+    } else {
+      return null;
+    }
+  }
+
+  private List<UserDefinedOutput> toCustomConfig(List<EventProperty> eventProperties) {
+    return eventProperties.stream().map(ep -> new UserDefinedOutput(
+        ep.getRuntimeName(),
+        ((EventPropertyPrimitive) ep).getRuntimeType(),
+        ep.getSemanticType())).toList();
+  }
+
   private String replaceId(String id) {
     return id.replaceAll(InvocablePipelineElementGenerator.ID_PREFIX, "");
   }

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/DataProcessorPipelineElementGenerator.java

@@ -35,7 +35,13 @@ public DataProcessorInvocation generate(DataProcessorInvocation processor,
                                           CompactPipelineElement pipelineElement) {
     basicGenerator.apply(processor, pipelineElement);
     var template = basicGenerator.makeTemplate(processor, pipelineElement);
-    return new DataProcessorTemplateHandler(template, processor, false)
+    var element = new DataProcessorTemplateHandler(template, processor, false)
         .applyTemplateOnPipelineElement();
+
+    if (pipelineElement.output() != null) {
+      var outputStrategyGenerator = new OutputStrategyGenerator(pipelineElement.output());
+      element.getOutputStrategies().forEach(o -> o.accept(outputStrategyGenerator));
+    }
+    return element;
   }
 }