Issue 695: Syslog stream returns message dataset instead of full dataset (#707)

* add CustomDataset object & test, set bloom create and update to return status dataset instead of the whole dataset.

* apply spotless

* add isStreaming(boolean) option for CustomDataset.

* separate makeRowsList and makeRowsSeq into Rows object & tests.

* change Bloom filter result to contain _time column and also switch result column to _raw.

* remove null sparksession from CustomDatasetTest.testEqualsContract

* remove double call to collectAsList() in CustomDatasetTest.testCreateStreamingDataset

* add CustomDataset interface, rename CustomDataset -> CustomDatasetImpl, remove isStreaming ctor arg, this is now handled by CustomResultStep object instead. Result dataset is now created in a separate CustomResultStep, which is added after bloom create and update.

* removed unused variables from TeragrepBloomStep

* change SyslogStreamer to use ForEachFunction instead of MapFunction, use Sequential mode, add custom result

* apply spotless

* fix TeragrepTest failing

* remove possibly unnecessary DataStreamWriter from CustomDatasetImpl

Author: eemhu

src/main/java/com/teragrep/pth_10/ast/commands/transformstatement/TeragrepTransformation.java

@@ -278,7 +278,16 @@ public Node visitT_syslogModeParameter(DPLParser.T_syslogModeParameterContext ct
             }
         }
 
-        return new StepNode(new TeragrepSyslogStep(host, port));
+        final CustomResultStep customResultStep = new CustomResultStep(
+                new CustomDatasetImpl(new StructType(new StructField[] {
+                        StructField.apply("_time", DataTypes.TimestampType, false, new MetadataBuilder().build()),
+                        StructField.apply("_raw", DataTypes.StringType, false, new MetadataBuilder().build())
+                }), Collections.singletonList(new Object[] {
+                        Instant.now(), "Syslog stream in progress..."
+                }), catCtx)
+        );
+
+        return new StepListNode(Arrays.asList(new TeragrepSyslogStep(host, port), customResultStep));
     }
 
     // exec hdfs save path retention

src/main/java/com/teragrep/pth_10/ast/commands/transformstatement/teragrep/SyslogStreamer.java

@@ -51,7 +51,7 @@
 import com.cloudbees.syslog.SyslogMessage;
 import com.teragrep.rlp_01.RelpBatch;
 import com.teragrep.rlp_01.RelpConnection;
-import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.api.java.function.ForeachFunction;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.types.StructField;
 import org.slf4j.Logger;
@@ -68,7 +68,7 @@
  * Streams the given dataset (using the map function of a dataset) as syslog messages via the RELP protocol.<br>
  * Provide the RELP server's hostname/ip and port using the constructor.
  */
-public class SyslogStreamer implements MapFunction<Row, Row>, Serializable {
+public class SyslogStreamer implements ForeachFunction<Row>, Serializable {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(SyslogStreamer.class);
     private transient RelpConnection sender;
@@ -235,10 +235,9 @@ private void stop() {
      * <br>
      *
      * @param row Input row to send as syslog
-     * @return the given input row unchanged
      * @throws Exception If any failure is encountered during the call
      */
-    public Row call(Row row) throws Exception {
+    public void call(Row row) throws Exception {
         if (!initialized) {
             this.start();
             initialized = true;
@@ -289,8 +288,6 @@ else if (field.name().equals("_raw")) {
 
         // send to server
         this.append(syslogMessage);
-
-        return row;
     }
 
     /**

src/main/java/com/teragrep/pth_10/steps/teragrep/TeragrepSyslogStep.java

@@ -67,14 +67,16 @@ public TeragrepSyslogStep(String relpHost, int relpPort) {
         this.relpHost = relpHost;
         this.relpPort = relpPort;
         this.properties.add(CommandProperty.NO_PRECEDING_AGGREGATE);
+        this.properties.add(CommandProperty.SEQUENTIAL_ONLY);
     }
 
     @Override
     public Dataset<Row> get(Dataset<Row> dataset) {
         LOGGER.info("Calling sendDataframeAsSyslog function");
         final SyslogStreamer syslogStreamer = new SyslogStreamer(relpHost, relpPort);
 
-        return dataset.map(syslogStreamer, dataset.exprEnc());
+        dataset.foreach(syslogStreamer);
+        return dataset;
     }
 
     @Override

src/test/java/com/teragrep/pth_10/translationTests/TeragrepTest.java

@@ -51,6 +51,7 @@
 import com.teragrep.pth_10.ast.bo.StepListNode;
 import com.teragrep.pth_10.ast.bo.StepNode;
 import com.teragrep.pth_10.ast.commands.transformstatement.TeragrepTransformation;
+import com.teragrep.pth_10.steps.CustomResultStep;
 import com.teragrep.pth_10.steps.teragrep.*;
 import com.teragrep.pth_03.antlr.DPLLexer;
 import com.teragrep.pth_03.antlr.DPLParser;
@@ -67,6 +68,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
+import java.util.List;
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 public class TeragrepTest {
@@ -86,15 +88,19 @@ void testTeragrepTranslation() {
         final DPLParserCatalystVisitor visitor = new DPLParserCatalystVisitor(ctx);
 
         final TeragrepTransformation ct = new TeragrepTransformation(ctx, visitor);
-        StepNode stepNode = (StepNode) ct
+        final StepListNode stepNode = (StepListNode) ct
                 .visitTeragrepTransformation((DPLParser.TeragrepTransformationContext) tree.getChild(1).getChild(0));
-        AbstractStep step = stepNode.get();
+        final List<AbstractStep> steps = stepNode.asList();
 
+        final AbstractStep step = steps.get(0);
         Assertions.assertEquals(TeragrepSyslogStep.class, step.getClass());
         TeragrepSyslogStep syslogStep = (TeragrepSyslogStep) step;
 
         Assertions.assertEquals("127.0.0.123", syslogStep.relpHost);
         Assertions.assertEquals(1337, syslogStep.relpPort);
+
+        final AbstractStep step2 = steps.get(1);
+        Assertions.assertEquals(CustomResultStep.class, step2.getClass());
     }
 
     @Test
@@ -110,15 +116,18 @@ void testTeragrepDefaultParamsTranslation() {
         final DPLParserCatalystVisitor visitor = new DPLParserCatalystVisitor(ctx);
 
         final TeragrepTransformation ct = new TeragrepTransformation(ctx, visitor);
-        StepNode stepNode = (StepNode) ct
+        final StepListNode stepNode = (StepListNode) ct
                 .visitTeragrepTransformation((DPLParser.TeragrepTransformationContext) tree.getChild(1).getChild(0));
-        final AbstractStep step = stepNode.get();
+        final List<AbstractStep> steps = stepNode.asList();
+        final AbstractStep step = steps.get(0);
 
         Assertions.assertEquals(TeragrepSyslogStep.class, step.getClass());
         final TeragrepSyslogStep syslogStep = (TeragrepSyslogStep) step;
 
         Assertions.assertEquals("127.0.0.1", syslogStep.relpHost);
         Assertions.assertEquals(601, syslogStep.relpPort);
+        final AbstractStep step2 = steps.get(1);
+        Assertions.assertEquals(CustomResultStep.class, step2.getClass());
     }
 
     @Test
@@ -144,15 +153,18 @@ void testTeragrepSyslogConfigTranslation() {
         final DPLParserCatalystVisitor visitor = new DPLParserCatalystVisitor(ctx);
 
         final TeragrepTransformation ct = new TeragrepTransformation(ctx, visitor);
-        StepNode stepNode = (StepNode) ct
+        final StepListNode stepNode = (StepListNode) ct
                 .visitTeragrepTransformation((DPLParser.TeragrepTransformationContext) tree.getChild(1).getChild(0));
-        final AbstractStep step = stepNode.get();
+        final List<AbstractStep> steps = stepNode.asList();
+        final AbstractStep step = steps.get(0);
 
         Assertions.assertEquals(TeragrepSyslogStep.class, step.getClass());
         final TeragrepSyslogStep syslogStep = (TeragrepSyslogStep) step;
 
         Assertions.assertEquals(host, syslogStep.relpHost);
         Assertions.assertEquals(port, syslogStep.relpPort);
+        final AbstractStep step2 = steps.get(1);
+        Assertions.assertEquals(CustomResultStep.class, step2.getClass());
     }
 
     @Test