Add post batchcollect mode and dedup implemented using flatMap (#636)

* move abstractStep to dpf_02, add aggregate after dedup test, add post batchcollect mode

* dedup: add DedupState state object, change DedupStep to use flatMapGroupsWithState, fix maxDuplicates default value, remove AbstractDedupStep

* convert internal batch collect commands to post batchcollect commands

* remove commented-out code from SortStep

* remove commented-out code from AddtotalsStep

* disable dedup consecutive parameter test as it is not implemented

* Upgrade JDK8 to JDK11

* remove unused batchcollects

* add 'dpl.pth_10.postbc.limit.size' config item to configure post-batchcollect limit size

* don't generate empty timechart buckets if post bc mode is used

* apply spotless

* Update dpf_02 to 4.0.0

---------

Co-authored-by: Mikko Kortelainen <[email protected]>

Author: eemhu

.github/workflows/ci.yaml

@@ -19,7 +19,7 @@ jobs:
       - name: Setup Java
         uses: actions/setup-java@v4
         with:
-          java-version: 8.0.292+10
+          java-version: 11
           distribution: 'adopt'
 
       - name: Install test dependencies

.github/workflows/coverity.yaml

@@ -25,7 +25,7 @@ jobs:
       - name: Setup Maven Central
         uses: actions/setup-java@v4
         with:
-          java-version: 8.0.292+10
+          java-version: 11
           distribution: 'adopt'
 
       - name: Cache Local Maven Repository

.github/workflows/upload_release_github_attachment.yaml

@@ -25,7 +25,7 @@ jobs:
       - name: Setup Java
         uses: actions/setup-java@v4
         with:
-          java-version: 8.0.292+10
+          java-version: 11
           distribution: 'adopt'
 
       - name: Install test dependencies

.github/workflows/upload_release_github_packages.yaml

@@ -25,13 +25,13 @@ jobs:
       - name: Setup Signing
         uses: actions/setup-java@v4
         with:
-          java-version: 8.0.292+10
+          java-version: 11
           distribution: 'adopt'
 
       - name: Setup GitHub Packages
         uses: actions/setup-java@v4
         with:
-          java-version: 8.0.292+10
+          java-version: 11
           distribution: 'adopt'
 
       - name: Install test dependencies

pom.xml

@@ -41,14 +41,14 @@
   </scm>
   <properties>
     <changelist>-SNAPSHOT</changelist>
-    <java.version>1.8</java.version>
+    <java.version>11</java.version>
     <junit.jupiter.version>5.7.1</junit.jupiter.version>
     <maven.compiler.source>1.8</maven.compiler.source>
     <maven.compiler.target>1.8</maven.compiler.target>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <revision>0.0.1</revision>
     <sha1></sha1>
-    <teragrep.dpf_02.version>3.0.0</teragrep.dpf_02.version>
+    <teragrep.dpf_02.version>4.0.0</teragrep.dpf_02.version>
     <teragrep.dpf_03.version>11.0.1</teragrep.dpf_03.version>
     <teragrep.jpr_01.version>3.1.1</teragrep.jpr_01.version>
     <teragrep.jue_01.version>0.4.3</teragrep.jue_01.version>
@@ -334,7 +334,7 @@
                   <version>3.2.5</version>
                 </requireMavenVersion>
                 <requireJavaVersion>
-                  <version>[1.8,1.9)</version>
+                  <version>[11,12)</version>
                 </requireJavaVersion>
                 <banDynamicVersions></banDynamicVersions>
                 <requirePluginVersions>
@@ -373,7 +373,7 @@
               <goal>jar</goal>
             </goals>
             <configuration>
-              <source>8</source>
+              <source>11</source>
             </configuration>
           </execution>
         </executions>
@@ -437,8 +437,8 @@
         <artifactId>maven-compiler-plugin</artifactId>
         <version>3.8.1</version>
         <configuration>
-          <source>8</source>
-          <target>8</target>
+          <source>11</source>
+          <target>11</target>
         </configuration>
       </plugin>
       <plugin>

src/main/java/com/teragrep/pth10/ast/DPLParserCatalystContext.java

@@ -45,7 +45,7 @@
  */
 package com.teragrep.pth10.ast;
 
-import com.teragrep.pth10.steps.AbstractStep;
+import com.teragrep.functions.dpf_02.AbstractStep;
 import com.teragrep.pth10.steps.Flushable;
 import com.typesafe.config.Config;
 import org.apache.spark.sql.Dataset;
@@ -209,6 +209,16 @@ public Integer getDplRecallSize() {
         return dplRecallSize;
     }
 
+    public Integer postBcLimitSize() {
+        if (config != null && config.hasPath("dpl.pth_10.postbc.limit.size")) {
+            return config.getInt("dpl.pth_10.postbc.limit.size");
+        }
+        else {
+            // default to no limit if no config available
+            return 0;
+        }
+    }
+
     private String baseUrl = null;
     private String paragraphUrl = null;
     private String notebookUrl = null;

src/main/java/com/teragrep/pth10/ast/DPLParserCatalystVisitor.java

@@ -45,13 +45,13 @@
  */
 package com.teragrep.pth10.ast;
 
+import com.teragrep.functions.dpf_02.AbstractStep;
 import com.teragrep.pth10.ast.bo.*;
 import com.teragrep.pth10.ast.bo.Token.Type;
 import com.teragrep.pth10.ast.commands.logicalstatement.LogicalStatementCatalyst;
 import com.teragrep.pth10.ast.commands.logicalstatement.LogicalStatementXML;
 import com.teragrep.pth10.ast.commands.logicalstatement.TimeStatement;
 import com.teragrep.pth10.ast.commands.transformstatement.TransformStatement;
-import com.teragrep.pth10.steps.AbstractStep;
 import com.teragrep.pth10.steps.EmptyDataframeStep;
 import com.teragrep.pth10.steps.logicalCatalyst.LogicalCatalystStep;
 import com.teragrep.pth10.steps.subsearch.SubsearchStep;

src/main/java/com/teragrep/pth10/ast/StepList.java

@@ -45,8 +45,8 @@
  */
 package com.teragrep.pth10.ast;
 
+import com.teragrep.functions.dpf_02.AbstractStep;
 import com.teragrep.functions.dpf_02.BatchCollect;
-import com.teragrep.pth10.steps.AbstractStep;
 import org.apache.spark.api.java.function.VoidFunction2;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
@@ -69,17 +69,20 @@
 
 public class StepList implements VoidFunction2<Dataset<Row>, Long> {
 
+    private enum BreakpointType {
+        SEQUENTIAL, POST_BC
+    }
+
     private static final Logger LOGGER = LoggerFactory.getLogger(StepList.class);
     private final List<AbstractStep> list;
-    private int breakpoint = -1;
+    private final Map<BreakpointType, Integer> breakpoints;
     private int aggregateCount = 0;
     private boolean useInternalBatchCollect = false;
     private boolean ignoreDefaultSorting = false;
 
     private OutputMode outputMode = OutputMode.Append();
     private Consumer<Dataset<Row>> batchHandler = null; // for UI
     private BatchCollect batchCollect; // standard batchCollect, used before sending batch event
-    private BatchCollect sequentialModeBatchCollect; // used if in append mode and in sequential, to allow aggregates in sequential mode
     private DPLParserCatalystVisitor catVisitor;
 
     public void setBatchCollect(BatchCollect batchCollect) {
@@ -90,6 +93,10 @@ public void setBatchHandler(Consumer<Dataset<Row>> batchHandler) {
         this.batchHandler = batchHandler;
     }
 
+    public BatchCollect batchCollect() {
+        return this.batchCollect;
+    }
+
     @Deprecated
     public Consumer<Dataset<Row>> getBatchHandler() {
         return batchHandler;
@@ -102,8 +109,12 @@ public void setCatVisitor(DPLParserCatalystVisitor catVisitor) {
     public StepList(DPLParserCatalystVisitor catVisitor) {
         this.list = new ArrayList<>();
         this.catVisitor = catVisitor;
-        this.batchCollect = new BatchCollect("_time", catVisitor.getCatalystContext().getDplRecallSize());
-        this.sequentialModeBatchCollect = new BatchCollect(null, catVisitor.getCatalystContext().getDplRecallSize());
+        this.batchCollect = new BatchCollect(
+                "_time",
+                catVisitor.getCatalystContext().getDplRecallSize(),
+                catVisitor.getCatalystContext().postBcLimitSize()
+        );
+        this.breakpoints = new HashMap<>();
     }
 
     /**
@@ -180,7 +191,11 @@ public Dataset<Row> executeSubsearch(Dataset<Row> ds) throws StreamingQueryExcep
     private DataStreamWriter<Row> executeFromStep(int fromStepIndex, Dataset<Row> ds) throws StreamingQueryException {
         for (int i = fromStepIndex; i < this.list.size(); i++) {
             AbstractStep step = this.list.get(i);
-            if (i == breakpoint) {
+            if (
+                (breakpoints.containsKey(BreakpointType.SEQUENTIAL) && breakpoints
+                        .get(BreakpointType.SEQUENTIAL) == i) || (breakpoints.containsKey(BreakpointType.POST_BC)
+                                && breakpoints.get(BreakpointType.POST_BC) == i)
+            ) {
                 // Switch to sequential; aka run the step inside forEachBatch
                 LOGGER.debug("breakpoint encountered at index <{}>", i);
 
@@ -193,12 +208,12 @@ private DataStreamWriter<Row> executeFromStep(int fromStepIndex, Dataset<Row> ds
     }
 
     private Dataset<Row> executeInBatch(Dataset<Row> ds) throws StreamingQueryException {
-        if (breakpoint == -1) { // no sequential ops
+        if (!breakpoints.containsKey(BreakpointType.SEQUENTIAL)) { // no sequential ops
             return ds;
         }
 
         // sequential ops found
-        for (int i = breakpoint; i < this.list.size(); i++) {
+        for (int i = breakpoints.get(BreakpointType.SEQUENTIAL); i < this.list.size(); i++) {
             AbstractStep step = this.list.get(i);
             LOGGER.info("Executing seq ops in batch: <{}>", step.toString());
             ds = step.get(ds);
@@ -216,7 +231,7 @@ private void analyze() {
 
         for (int i = 0; i < this.list.size(); i++) {
             AbstractStep step = this.list.get(i);
-
+            LOGGER.info("Analyzing step: <{}>", step.toString());
             step.setAggregatesUsedBefore(aggregateCount > 0);
 
             if (step.hasProperty(AbstractStep.CommandProperty.USES_INTERNAL_BATCHCOLLECT)) {
@@ -228,7 +243,11 @@ private void analyze() {
             if (step.hasProperty(AbstractStep.CommandProperty.IGNORE_DEFAULT_SORTING)) {
                 LOGGER.info("[Analyze] Ignore default sorting: <{}>", step);
                 this.ignoreDefaultSorting = true;
-                this.batchCollect = new BatchCollect(null, catVisitor.getDPLRecallSize());
+                this.batchCollect = new BatchCollect(
+                        null,
+                        catVisitor.getDPLRecallSize(),
+                        catVisitor.getCatalystContext().postBcLimitSize()
+                );
             }
 
             if (step.hasProperty(AbstractStep.CommandProperty.REQUIRE_PRECEDING_AGGREGATE)) {
@@ -246,20 +265,29 @@ private void analyze() {
             if (step.hasProperty(AbstractStep.CommandProperty.SEQUENTIAL_ONLY)) {
                 LOGGER.info("[Analyze] Sequential only command: <{}>", step);
                 // set the breakpoint just once
-                if (breakpoint == -1) {
-                    breakpoint = i;
+                if (!breakpoints.containsKey(BreakpointType.SEQUENTIAL)) {
+                    breakpoints.put(BreakpointType.SEQUENTIAL, i);
                 }
             }
             else if (step.hasProperty(AbstractStep.CommandProperty.AGGREGATE)) {
                 LOGGER.info("[Analyze] Aggregate command: <{}>", step);
                 aggregateCount++;
 
                 // set the breakpoint just once
-                if (aggregateCount > 0 && breakpoint == -1) {
-                    breakpoint = i + 1;
+                if (
+                    aggregateCount > 0 && !breakpoints.containsKey(BreakpointType.SEQUENTIAL)
+                            && !breakpoints.containsKey(BreakpointType.POST_BC)
+                ) {
+                    breakpoints.put(BreakpointType.SEQUENTIAL, i + 1);
                     outputMode = OutputMode.Complete();
                 }
             }
+            else if (step.hasProperty(AbstractStep.CommandProperty.POST_BATCHCOLLECT)) {
+                if (!breakpoints.containsKey(BreakpointType.POST_BC)) {
+                    LOGGER.info("[Analyze] Post batch collect command: <{}>", step);
+                    breakpoints.put(BreakpointType.POST_BC, i);
+                }
+            }
         }
     }
 
@@ -282,7 +310,11 @@ else if (this.batchCollect == null) {
             }
             else {
                 LOGGER.info("------------------ Aggregates NOT USED (before seq. switch), using batchCollect!");
-                this.batchCollect.collect(ds, id);
+                int index = this.list.size();
+                if (breakpoints.containsKey(BreakpointType.POST_BC)) {
+                    index = breakpoints.get(BreakpointType.POST_BC);
+                }
+                this.batchCollect.collect(ds, id, this.list.subList(index, this.list.size()), false);
                 this.batchHandler.accept(batchCollect.getCollectedAsDataframe());
             }
         }
@@ -292,7 +324,10 @@ public void call(Dataset<Row> batchDF, Long batchId) throws StreamingQueryExcept
         LOGGER.info("StepList batch processing received a new batch <{}>", batchId);
 
         // timechart empty buckets
-        if (catVisitor.getCatalystContext().getTimeChartSpanSeconds() != null) {
+        if (
+            catVisitor.getCatalystContext().getTimeChartSpanSeconds() != null
+                    && !breakpoints.containsKey(BreakpointType.POST_BC)
+        ) {
             // create spans
             final long min = catVisitor.getCatalystContext().getDplMinimumEarliest();
             final long max = catVisitor.getCatalystContext().getDplMaximumLatest();
@@ -346,11 +381,7 @@ else if (dataType == DataTypes.FloatType) {
 
         // Continue sub list of steps execution, if necessary
         if (!this.list.isEmpty()) {
-            LOGGER
-                    .info(
-                            "StepList batch processing - Continuing execution to next ops after breakpoint index: <{}>",
-                            breakpoint
-                    );
+            LOGGER.info("StepList batch processing - Continuing execution to next ops after breakpoint index");
 
             Dataset<Row> ret = this.executeInBatch(batchDF);
 

src/main/java/com/teragrep/pth10/ast/bo/StepListNode.java

@@ -45,7 +45,7 @@
  */
 package com.teragrep.pth10.ast.bo;
 
-import com.teragrep.pth10.steps.AbstractStep;
+import com.teragrep.functions.dpf_02.AbstractStep;
 
 import java.util.ArrayList;
 import java.util.List;

src/main/java/com/teragrep/pth10/ast/bo/StepNode.java

@@ -45,7 +45,7 @@
  */
 package com.teragrep.pth10.ast.bo;
 
-import com.teragrep.pth10.steps.AbstractStep;
+import com.teragrep.functions.dpf_02.AbstractStep;
 
 public class StepNode extends Node {