Add pth_15 class implementations, move pth_07 DPLExecutor stuff here (#657)

* add pth_15 class implementations, move DPLExecutor from zep_01/pth_07 to pth_10, change BatchHandler type to BiConsumer, add pth_15 dependency.

* remove runIncrement and queryId fields from DPLExecutorImpl; remove metrics from DPLExecutorResultImpl; remove consumers from DPLStreamingQueryListener; update pth_15 to 2.0.0

Author: eemhu

pom.xml

@@ -54,10 +54,16 @@
     <teragrep.jue_01.version>0.4.3</teragrep.jue_01.version>
     <teragrep.pth_03.version>9.3.0</teragrep.pth_03.version>
     <teragrep.pth_06.version>3.5.0</teragrep.pth_06.version>
+    <teragrep.pth_15.version>2.0.0</teragrep.pth_15.version>
     <teragrep.rlp_01.version>4.0.1</teragrep.rlp_01.version>
     <teragrep.rlp_03.version>9.0.0</teragrep.rlp_03.version>
   </properties>
   <dependencies>
+    <dependency>
+      <groupId>com.teragrep</groupId>
+      <artifactId>pth_15</artifactId>
+      <version>${teragrep.pth_15.version}</version>
+    </dependency>
     <!--DPL-dependencies -->
     <dependency>
       <groupId>com.teragrep</groupId>

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

@@ -69,6 +69,7 @@
 import javax.xml.parsers.DocumentBuilderFactory;
 import javax.xml.parsers.ParserConfigurationException;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 /**
@@ -124,7 +125,7 @@ public List<String> getTraceBuffer() {
     }
 
     @Deprecated
-    public Consumer<Dataset<Row>> getConsumer() {
+    public BiConsumer<Dataset<Row>, Boolean> getConsumer() {
         return this.stepList.getBatchHandler();
     }
 
@@ -133,7 +134,7 @@ public Consumer<Dataset<Row>> getConsumer() {
      * 
      * @param consumer Consumer with type Dataset to be implemented in pth_07
      */
-    public void setConsumer(Consumer<Dataset<Row>> consumer) {
+    public void setConsumer(BiConsumer<Dataset<Row>, Boolean> consumer) {
         this.stepList.setBatchHandler(consumer);
     }
 

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

@@ -65,7 +65,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 public class StepList implements VoidFunction2<Dataset<Row>, Long> {
 
@@ -81,15 +81,15 @@ private enum BreakpointType {
     private boolean ignoreDefaultSorting = false;
 
     private OutputMode outputMode = OutputMode.Append();
-    private Consumer<Dataset<Row>> batchHandler = null; // for UI
+    private BiConsumer<Dataset<Row>, Boolean> batchHandler = null; // for UI
     private BatchCollect batchCollect; // standard batchCollect, used before sending batch event
     private DPLParserCatalystVisitor catVisitor;
 
     public void setBatchCollect(BatchCollect batchCollect) {
         this.batchCollect = batchCollect;
     }
 
-    public void setBatchHandler(Consumer<Dataset<Row>> batchHandler) {
+    public void setBatchHandler(BiConsumer<Dataset<Row>, Boolean> batchHandler) {
         this.batchHandler = batchHandler;
     }
 
@@ -98,7 +98,7 @@ public BatchCollect batchCollect() {
     }
 
     @Deprecated
-    public Consumer<Dataset<Row>> getBatchHandler() {
+    public BiConsumer<Dataset<Row>, Boolean> getBatchHandler() {
         return batchHandler;
     }
 
@@ -299,14 +299,15 @@ else if (step.hasProperty(AbstractStep.CommandProperty.POST_BATCHCOLLECT)) {
      * @param id ID of the processed batch dataset
      */
     private void sendBatchEvent(Dataset<Row> ds, Long id) {
+        boolean aggregatesUsed = aggregateCount > 0;
         if (this.batchHandler != null) {
             if (outputMode == OutputMode.Complete()) {
                 LOGGER.info("------------------ Aggregates (Complete Mode) used, sending batch event!");
-                this.batchHandler.accept(ds);
+                this.batchHandler.accept(ds, aggregatesUsed);
             }
             else if (this.batchCollect == null) {
                 LOGGER.info("------------------ No batchCollect present (no sorting column), sending batch event!");
-                this.batchHandler.accept(ds);
+                this.batchHandler.accept(ds, aggregatesUsed);
             }
             else {
                 LOGGER.info("------------------ Aggregates NOT USED (before seq. switch), using batchCollect!");
@@ -315,7 +316,7 @@ else if (this.batchCollect == null) {
                     index = breakpoints.get(BreakpointType.POST_BC);
                 }
                 this.batchCollect.collect(ds, id, this.list.subList(index, this.list.size()), false);
-                this.batchHandler.accept(batchCollect.getCollectedAsDataframe());
+                this.batchHandler.accept(batchCollect.getCollectedAsDataframe(), aggregatesUsed);
             }
         }
     }

src/main/java/com/teragrep/pth_10/executor/DPLErrorListenerImpl.java

@@ -0,0 +1,87 @@
+/*
+ * Teragrep Data Processing Language (DPL) translator for Apache Spark (pth_10)
+ * Copyright (C) 2019-2025 Suomen Kanuuna Oy
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ *
+ *
+ * Additional permission under GNU Affero General Public License version 3
+ * section 7
+ *
+ * If you modify this Program, or any covered work, by linking or combining it
+ * with other code, such other code is not for that reason alone subject to any
+ * of the requirements of the GNU Affero GPL version 3 as long as this Program
+ * is the same Program as licensed from Suomen Kanuuna Oy without any additional
+ * modifications.
+ *
+ * Supplemented terms under GNU Affero General Public License version 3
+ * section 7
+ *
+ * Origin of the software must be attributed to Suomen Kanuuna Oy. Any modified
+ * versions must be marked as "Modified version of" The Program.
+ *
+ * Names of the licensors and authors may not be used for publicity purposes.
+ *
+ * No rights are granted for use of trade names, trademarks, or service marks
+ * which are in The Program if any.
+ *
+ * Licensee must indemnify licensors and authors for any liability that these
+ * contractual assumptions impose on licensors and authors.
+ *
+ * To the extent this program is licensed as part of the Commercial versions of
+ * Teragrep, the applicable Commercial License may apply to this file if you as
+ * a licensee so wish it.
+ */
+package com.teragrep.pth_10.executor;
+
+import com.teragrep.pth_03.shaded.org.antlr.v4.runtime.BaseErrorListener;
+import com.teragrep.pth_03.shaded.org.antlr.v4.runtime.RecognitionException;
+import com.teragrep.pth_03.shaded.org.antlr.v4.runtime.Recognizer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used to listen for errors in DPL queries. Overrides empty functions from BaseErrorListener.
+ */
+public final class DPLErrorListenerImpl extends BaseErrorListener {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DPLErrorListenerImpl.class);
+
+    private final String listenedTo;
+
+    public DPLErrorListenerImpl(String listenedTo) {
+        this.listenedTo = listenedTo;
+    }
+
+    @Override
+    public void syntaxError(
+            Recognizer<?, ?> recognizer,
+            Object offendingSymbol,
+            int line,
+            int charPositionInLine,
+            String msg,
+            RecognitionException e
+    ) {
+        if (e == null) {
+            LOGGER.error("Got an exception from <{}>, no message", listenedTo);
+        }
+        else {
+            LOGGER.error("Got an exception from <{}>: <[{}]>", listenedTo, e.getMessage(), e);
+        }
+        throw new IllegalStateException(
+                listenedTo + " failure on line " + line + ", column " + charPositionInLine + " due to " + msg + "\n"
+                        + "Please check that the query is written correctly. Otherwise, please report this error and include the query used and this error."
+        );
+    }
+}