Beam SQL UDF cleanup

nielsbasjes · Sep 26, 2021 · 0f37571 · 0f37571
1 parent 1450e08
commit 0f37571
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 7 deletions.
diff --git a/src/main/docs/UDF-ApacheBeamSql.md b/src/main/docs/UDF-ApacheBeamSql.md
@@ -30,7 +30,7 @@ to give
 
     Phone
 
-### Getting several values as a Map (requires Apache Beam 2.34.0 or newer)
+### Getting several values as a Map (requires Apache Beam 2.30.0 or newer)
 You can ask for a all fields and return the full map with all of them in there.
 
     ParseUserAgent(userAgent)                                    AS allFields

diff --git a/udfs/beam-sql/src/main/java/nl/basjes/parse/useragent/beam/BaseParseUserAgentUDF.java b/udfs/beam-sql/src/main/java/nl/basjes/parse/useragent/beam/BaseParseUserAgentUDF.java
@@ -24,6 +24,9 @@ abstract class BaseParseUserAgentUDF implements BeamSqlUdf {
     private static transient UserAgentAnalyzer userAgentAnalyzer = null;
 
     protected static UserAgentAnalyzer getInstance() {
+        // NOTE: We currently do NOT make an instance with only the wanted fields.
+        //       We only know the required parameters the moment the call is done.
+        //       At that point it is too late to create an optimized instance.
         if (userAgentAnalyzer == null) {
             userAgentAnalyzer = UserAgentAnalyzer
                 .newBuilder()

diff --git a/udfs/beam-sql/src/test/java/nl/basjes/parse/useragent/beam/TestParseUserAgentSQL.java b/udfs/beam-sql/src/test/java/nl/basjes/parse/useragent/beam/TestParseUserAgentSQL.java
@@ -43,6 +43,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 
 @Category(ValidatesRunner.class)
 public class TestParseUserAgentSQL implements Serializable {
@@ -308,9 +309,10 @@ public void testUserAgentAnalysisSQL() { // NOSONAR java:S2699 Tests should incl
             .apply(Create.of(useragents))
             .setCoder(ListCoder.of(StringUtf8Coder.of()))
             .apply(ParDo.of(new DoFn<List<String>, Row>() {
+                @SuppressWarnings("unused") // Called via the annotation
                 @ProcessElement
                 public void processElement(ProcessContext c) {
-                    c.output(listToRow(c.element(), inputSchema));
+                    c.output(listToRow(Objects.requireNonNull(c.element()), inputSchema));
                 }
             }))
             .setCoder(RowCoder.of(inputSchema));
@@ -431,7 +433,7 @@ public void processElement(ProcessContext c) {
                 );
 
         // Just to see the output of the query while debugging
-//        result.apply(ParDo.of(new RowPrinter()));
+        result.apply(ParDo.of(new RowPrinter()));
 
         // Assert on the results.
         PAssert.that(result)
@@ -441,7 +443,7 @@ public void processElement(ProcessContext c) {
     }
 
     public static class RowPrinter extends DoFn<Row, Void> {
-        @SuppressWarnings("unused") // Used via reflection
+        @SuppressWarnings("unused") // Called via the annotation
         @ProcessElement
         public void processElement(ProcessContext c) {
             final Row row = c.element();

diff --git a/udfs/beam-sql/src/test/java/nl/basjes/parse/useragent/beam/TestParseUserAgentSQLMap.java b/udfs/beam-sql/src/test/java/nl/basjes/parse/useragent/beam/TestParseUserAgentSQLMap.java
@@ -32,7 +32,6 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -47,7 +46,6 @@
 
 import static nl.basjes.parse.useragent.UserAgent.USERAGENT_FIELDNAME;
 
-@Ignore("This test will fail on Beam before 2.34.0 because of a needed fix.")
 @Category(ValidatesRunner.class)
 public class TestParseUserAgentSQLMap implements Serializable {
 
@@ -236,7 +234,7 @@ public void testUserAgentAnalysisSQL() { // NOSONAR java:S2699 Tests should incl
     }
 
     public static class RowPrinter extends DoFn<Row, Void> {
-        @SuppressWarnings("unused") // Used via reflection
+        @SuppressWarnings("unused") // Called via the annotation
         @ProcessElement
         public void processElement(ProcessContext c) {
             final Row row = c.element();