Issue 647: Bloom create and update now returns only status messages. (#693)

eemhu · web-flow · commit 6ce5255ee252 · 2025-11-11T15:17:58.000+02:00
* add CustomDataset object &amp; 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 &amp; 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 -&gt; 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

* remove possibly unnecessary DataStreamWriter from CustomDatasetImpl
diff --git a/src/main/java/com/teragrep/pth_10/ast/commands/transformstatement/TeragrepTransformation.java b/src/main/java/com/teragrep/pth_10/ast/commands/transformstatement/TeragrepTransformation.java
@@ -61,6 +61,8 @@
 import com.teragrep.pth_10.ast.commands.transformstatement.teragrep.RegexValueFromBloomContext;
 import com.teragrep.pth_10.ast.commands.transformstatement.teragrep.TableNameFromBloomContext;
 import com.teragrep.pth_10.ast.ContextValue;
+import com.teragrep.pth_10.datasources.CustomDatasetImpl;
+import com.teragrep.pth_10.steps.CustomResultStep;
 import com.teragrep.pth_10.steps.teragrep.*;
 import com.teragrep.pth_10.steps.teragrep.AbstractTokenizerStep;
 import com.teragrep.pth_10.steps.teragrep.TeragrepTokenizerStep;
@@ -69,13 +71,19 @@
 import com.teragrep.pth_03.antlr.DPLParser;
 import com.teragrep.pth_03.antlr.DPLParserBaseVisitor;
 import com.teragrep.pth_03.shaded.org.antlr.v4.runtime.tree.TerminalNode;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.MetadataBuilder;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.w3c.dom.Document;
 
 import javax.xml.parsers.DocumentBuilderFactory;
 import javax.xml.parsers.ParserConfigurationException;
+import java.time.Instant;
 import java.util.Arrays;
+import java.util.Collections;
 
 /**
  * Class containing the visitor methods for all "| teragrep" subcommands
@@ -509,7 +517,18 @@ public Node visitT_bloomOptionParameter(final DPLParser.T_bloomOptionParameterCo
                     outputCol.value(),
                     estimateCol.value()
             );
-            rv = new StepListNode(Arrays.asList(aggregateStep, bloomStepWithRegexAndTable));
+            // Create a step, that returns a Custom dataset containing the result message
+            // instead of the whole dataset
+            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(),
+                            "Bloom filter operation ".concat(mode.value().toString()).concat(" was completed.")
+                    }), catCtx)
+            );
+            rv = new StepListNode(Arrays.asList(aggregateStep, bloomStepWithRegexAndTable, customResultStep));
         }
         else {
             final TeragrepBloomStep bloomStep = new TeragrepBloomStep(
diff --git a/src/main/java/com/teragrep/pth_10/datasources/CustomDataset.java b/src/main/java/com/teragrep/pth_10/datasources/CustomDataset.java
@@ -0,0 +1,56 @@
+/*
+ * 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.datasources;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+
+public interface CustomDataset {
+
+    public abstract Dataset<Row> asStaticDataset();
+
+    public abstract Dataset<Row> asStreamingDataset();
+}
diff --git a/src/main/java/com/teragrep/pth_10/datasources/CustomDatasetImpl.java b/src/main/java/com/teragrep/pth_10/datasources/CustomDatasetImpl.java
@@ -0,0 +1,105 @@
+/*
+ * 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.datasources;
+
+import com.teragrep.pth_10.ast.DPLParserCatalystContext;
+import org.apache.spark.sql.*;
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder;
+import org.apache.spark.sql.catalyst.encoders.RowEncoder;
+import org.apache.spark.sql.execution.streaming.MemoryStream;
+import org.apache.spark.sql.types.StructType;
+import scala.Option;
+
+import java.util.List;
+import java.util.Objects;
+
+public final class CustomDatasetImpl implements CustomDataset {
+
+    private final StructType schema;
+    private final List<Object[]> values;
+    private final DPLParserCatalystContext catCtx;
+
+    public CustomDatasetImpl(
+            final StructType schema,
+            final List<Object[]> values,
+            final DPLParserCatalystContext catCtx
+    ) {
+        this.schema = schema;
+        this.values = values;
+        this.catCtx = catCtx;
+    }
+
+    @Override
+    public Dataset<Row> asStaticDataset() {
+        return catCtx.getSparkSession().createDataFrame(new Rows(values).asList(), schema);
+    }
+
+    @Override
+    public Dataset<Row> asStreamingDataset() {
+        final SQLContext sqlContext = catCtx.getSparkSession().sqlContext();
+        final ExpressionEncoder<Row> encoder = RowEncoder.apply(schema);
+        final MemoryStream<Row> rowMemoryStream = new MemoryStream<>(1, sqlContext, Option.apply(1), encoder);
+
+        rowMemoryStream.addData(new Rows(values).asSeq());
+
+        return rowMemoryStream.toDF();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final CustomDatasetImpl that = (CustomDatasetImpl) o;
+        return Objects.equals(schema, that.schema) && Objects.equals(values, that.values)
+                && Objects.equals(catCtx, that.catCtx);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(schema, values, catCtx);
+    }
+}
diff --git a/src/main/java/com/teragrep/pth_10/datasources/Rows.java b/src/main/java/com/teragrep/pth_10/datasources/Rows.java
@@ -0,0 +1,90 @@
+/*
+ * 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.datasources;
+
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+public final class Rows {
+
+    private final List<Object[]> values;
+
+    public Rows(final List<Object[]> values) {
+        this.values = values;
+    }
+
+    public Seq<Row> asSeq() {
+        return JavaConverters.asScalaBuffer(asList()).toSeq();
+    }
+
+    public List<Row> asList() {
+        final List<Row> rows = new ArrayList<>();
+        for (final Object[] rowValues : values) {
+            rows.add(RowFactory.create(rowValues));
+        }
+        return rows;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final Rows rows = (Rows) o;
+        return Objects.equals(values, rows.values);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(values);
+    }
+}
diff --git a/src/main/java/com/teragrep/pth_10/steps/CustomResultStep.java b/src/main/java/com/teragrep/pth_10/steps/CustomResultStep.java
@@ -0,0 +1,75 @@
+/*
+ * 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.steps;
+
+import com.teragrep.functions.dpf_02.AbstractStep;
+import com.teragrep.pth_10.datasources.CustomDataset;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.streaming.StreamingQueryException;
+
+public final class CustomResultStep extends AbstractStep {
+
+    private final CustomDataset customDataset;
+
+    public CustomResultStep(final CustomDataset customDataset) {
+        this.customDataset = customDataset;
+    }
+
+    @Override
+    public Dataset<Row> get(final Dataset<Row> dataset) throws StreamingQueryException {
+        final Dataset<Row> result;
+
+        if (dataset.isStreaming()) {
+            result = customDataset.asStreamingDataset();
+        }
+        else {
+            result = customDataset.asStaticDataset();
+        }
+
+        return result;
+    }
+}
diff --git a/src/main/java/com/teragrep/pth_10/steps/teragrep/TeragrepBloomStep.java b/src/main/java/com/teragrep/pth_10/steps/teragrep/TeragrepBloomStep.java
@@ -55,8 +55,6 @@
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.functions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * teragrep exec bloom
@@ -67,8 +65,6 @@ public enum BloomMode {
         UPDATE, CREATE, ESTIMATE, AGGREGATE, DEFAULT
     }
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(TeragrepBloomStep.class);
-
     private final Config zeppelinConfig;
     public final BloomMode mode;
     private final String tableName;
diff --git a/src/test/java/com/teragrep/pth_10/CustomDatasetImplTest.java b/src/test/java/com/teragrep/pth_10/CustomDatasetImplTest.java
diff --git a/src/test/java/com/teragrep/pth_10/RowsTest.java b/src/test/java/com/teragrep/pth_10/RowsTest.java
