feat: organize UDFs by category (#5813)

A little friday afternoon, quality-of-life-improvement: i noticed that we now have quite some UDFs, which is cool, but the list you get when executing show functions is quite long and hard to navigate unless you really know what you are looking for. Hence this PR adds a new categorization of UDFs into STRING|AGGREGATE|ARRAY|... etc.

Also a small message at the end of the listing detailing how to get further information about a function of interest.

docs/concepts/functions.md

@@ -258,6 +258,7 @@ commands.
 |-------------|----------------------------------------------------------------------|----------|
 | name        | The case-insensitive name of the UDF(s) represented by this class.   | Yes      |
 | description | A string describing generally what the function(s) in this class do. | Yes      |
+| category    | For grouping similar functions in the output of SHOW FUNCTIONS.      | No       |
 | author      | The author of the UDF.                                               | No       |
 | version     | The version of the UDF.                                              | No       |
 

ksqldb-cli/src/main/java/io/confluent/ksql/cli/console/table/builder/FunctionNameListTableBuilder.java

@@ -1,16 +1,15 @@
 /*
  * Copyright 2018 Confluent Inc.
  *
- * Licensed under the Confluent Community License (the "License"); you may not use
- * this file except in compliance with the License.  You may obtain a copy of the
- * License at
+ * Licensed under the Confluent Community License (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
  *
  * http://www.confluent.io/confluent-community-license
  *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the
+ * License.
  */
 
 package io.confluent.ksql.cli.console.table.builder;
@@ -19,24 +18,47 @@
 import io.confluent.ksql.cli.console.table.Table;
 import io.confluent.ksql.cli.console.table.Table.Builder;
 import io.confluent.ksql.rest.entity.FunctionNameList;
+import io.confluent.ksql.rest.entity.SimpleFunctionInfo;
+import java.util.Comparator;
+import java.util.Iterator;
 import java.util.List;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 public class FunctionNameListTableBuilder implements TableBuilder<FunctionNameList> {
 
-  private static final List<String> HEADERS =
-      ImmutableList.of("Function Name", "Type");
+  private static final List<String> HEADERS = ImmutableList.of("Function Name", "Category");
+  private static final List<String> EMPTY_ROW = ImmutableList.of(" ", " ");
 
   @Override
   public Table buildTable(final FunctionNameList functionNameList) {
-    final Stream<List<String>> rows = functionNameList.getFunctions()
-        .stream()
-        .sorted()
-        .map(func -> ImmutableList.of(func.getName(), func.getType().name().toUpperCase()));
 
-    return new Builder()
-        .withColumnHeaders(HEADERS)
-        .withRows(rows)
-        .build();
+    final Builder builder = new Builder().withColumnHeaders(HEADERS);
+
+    // poor mans version check for the case we are running against an older ksqlDB server
+    final Iterator<SimpleFunctionInfo> funcs = functionNameList.getFunctions().iterator();
+    if (!funcs.hasNext() || funcs.next().getCategory().isEmpty()) {
+      final Stream<List<String>> rows = functionNameList.getFunctions()
+          .stream()
+          .sorted(Comparator.comparing(SimpleFunctionInfo::getName))
+          .map(func -> ImmutableList.of(func.getName(), func.getType().name().toUpperCase()));
+      builder.withRows(rows);
+    } else { // category info present, use new display layout
+      final List<SimpleFunctionInfo> sortedFunctions = functionNameList.getFunctions().stream()
+          .sorted(Comparator.comparing(SimpleFunctionInfo::getCategory)
+              .thenComparing(SimpleFunctionInfo::getName))
+          .collect(Collectors.toList());
+      String prevCategory = sortedFunctions.get(0).getCategory();
+      for (SimpleFunctionInfo fn : sortedFunctions) {
+        if (!fn.getCategory().equals(prevCategory)) {
+          builder.withRow(EMPTY_ROW);
+        }
+        builder.withRow(fn.getName(), fn.getCategory());
+        prevCategory = fn.getCategory();
+      }
+    }
+    builder.withFooterLine(
+        "For detailed information about a function, run: DESCRIBE FUNCTION <Function Name>;");
+    return builder.build();
   }
 }

ksqldb-cli/src/test/java/io/confluent/ksql/cli/CliTest.java

@@ -784,8 +784,8 @@ public void shouldPrintErrorOnUnsupportedAPI() throws Exception {
   @Test
   public void shouldListFunctions() {
     assertRunListCommand("functions", hasRows(
-        row("TIMESTAMPTOSTRING", "SCALAR"),
-        row("EXTRACTJSONFIELD", "SCALAR"),
+        row("TIMESTAMPTOSTRING", "DATE / TIME"),
+        row("EXTRACTJSONFIELD", "JSON"),
         row("TOPK", "AGGREGATE")
     ));
   }

ksqldb-common/src/main/java/io/confluent/ksql/function/AggregateFunctionFactory.java

@@ -49,6 +49,7 @@ public AggregateFunctionFactory(final String functionName) {
         "",
         KsqlConstants.CONFLUENT_AUTHOR,
         "",
+        FunctionCategory.AGGREGATE,
         KsqlScalarFunction.INTERNAL_PATH
     ));
   }

ksqldb-common/src/main/java/io/confluent/ksql/function/udf/UdfMetadata.java

@@ -23,17 +23,20 @@ public class UdfMetadata {
   private final String author;
   private final String version;
   private final String path;
+  private final String category;
 
   public UdfMetadata(final String name,
                      final String description,
                      final String author,
                      final String version,
+      final String category,
                      final String path
   ) {
     this.name = Objects.requireNonNull(name, "name cant be null");
     this.description = Objects.requireNonNull(description, "description can't be null");
     this.author = Objects.requireNonNull(author, "author can't be null");
     this.version = Objects.requireNonNull(version, "version can't be null");
+    this.category = Objects.requireNonNull(category, "category can't be null");
     this.path = Objects.requireNonNull(path, "path can't be null");
   }
 
@@ -57,14 +60,19 @@ public String getPath() {
     return path;
   }
 
+  public String getCategory() {
+    return category;
+  }
+
   @Override
   public String toString() {
     return "UdfMetadata{"
         + "name='" + name + '\''
         + ", description='" + description + '\''
         + ", author='" + author + '\''
         + ", version='" + version + '\''
-        + ", path='" + path + "'"
+        + ", path='" + path + '\''
+        + ", category='" + category + "'"
         + '}';
   }
 
@@ -81,11 +89,12 @@ public boolean equals(final Object o) {
         && Objects.equals(description, that.description)
         && Objects.equals(author, that.author)
         && Objects.equals(version, that.version)
-        && Objects.equals(path, that.path);
+        && Objects.equals(path, that.path)
+        && Objects.equals(category, that.category);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(name, description, author, version, path);
+    return Objects.hash(name, description, author, version, path, category);
   }
 }

ksqldb-common/src/test/java/io/confluent/ksql/function/UdfFactoryTest.java

@@ -35,7 +35,7 @@ public class UdfFactoryTest {
 
   private static final String functionName = "TestFunc";
   private final UdfFactory factory = new UdfFactory(TestFunc.class,
-      new UdfMetadata(functionName, "", "", "", "internal"));
+      new UdfMetadata(functionName, "", "", "", FunctionCategory.OTHER, "internal"));
 
   @Test
   public void shouldThrowIfNoVariantFoundThatAcceptsSuppliedParamTypes() {

ksqldb-common/src/test/java/io/confluent/ksql/function/udf/UdfMetadataTest.java

@@ -1,20 +1,22 @@
 /*
  * Copyright 2018 Confluent Inc.
  *
- * Licensed under the Confluent Community License (the "License"); you may not use
- * this file except in compliance with the License.  You may obtain a copy of the
- * License at
+ * Licensed under the Confluent Community License (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
  *
  * http://www.confluent.io/confluent-community-license
  *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the
+ * License.
  */
 
 package io.confluent.ksql.function.udf;
 
+import static io.confluent.ksql.function.FunctionCategory.MAP;
+import static io.confluent.ksql.function.FunctionCategory.OTHER;
+
 import com.google.common.testing.EqualsTester;
 import org.junit.Test;
 
@@ -24,14 +26,14 @@ public class UdfMetadataTest {
   public void shouldImplementHashCodeAndEqualsProperly() {
     new EqualsTester()
         .addEqualityGroup(
-            new UdfMetadata("name", "desc", "auth", "ver", "path"),
-            new UdfMetadata("name", "desc", "auth", "ver", "path")
-        )
-        .addEqualityGroup(new UdfMetadata("DIF", "desc", "auth", "ver", "path"))
-        .addEqualityGroup(new UdfMetadata("name", "DIF", "auth", "ver", "path"))
-        .addEqualityGroup(new UdfMetadata("name", "desc", "DIF", "ver", "path"))
-        .addEqualityGroup(new UdfMetadata("name", "desc", "auth", "DIF", "path"))
-        .addEqualityGroup(new UdfMetadata("name", "desc", "auth", "ver", "DIF"))
+            new UdfMetadata("name", "desc", "auth", "ver", OTHER, "path"),
+            new UdfMetadata("name", "desc", "auth", "ver", OTHER, "path"))
+        .addEqualityGroup(new UdfMetadata("DIF", "desc", "auth", "ver", OTHER, "path"))
+        .addEqualityGroup(new UdfMetadata("name", "DIF", "auth", "ver", OTHER, "path"))
+        .addEqualityGroup(new UdfMetadata("name", "desc", "DIF", "ver", OTHER, "path"))
+        .addEqualityGroup(new UdfMetadata("name", "desc", "auth", "DIF", OTHER, "path"))
+        .addEqualityGroup(new UdfMetadata("name", "desc", "auth", "ver", MAP, "path"))
+        .addEqualityGroup(new UdfMetadata("name", "desc", "auth", "ver", OTHER, "DIF"))
         .testEquals();
   }
-}
+}

ksqldb-engine/src/main/java/io/confluent/ksql/function/UdafLoader.java

@@ -102,6 +102,7 @@ void loadUdafFromClass(final Class<?> theClass, final String path) {
             udafAnnotation.description(),
             udafAnnotation.author(),
             udafAnnotation.version(),
+            udafAnnotation.category(),
             path
         ),
         invokers

ksqldb-engine/src/main/java/io/confluent/ksql/function/UdfLoader.java

@@ -93,6 +93,7 @@ public void loadUdfFromClass(
             udfDescriptionAnnotation.description(),
             udfDescriptionAnnotation.author(),
             udfDescriptionAnnotation.version(),
+            udfDescriptionAnnotation.category(),
             path
         )
     );

ksqldb-engine/src/main/java/io/confluent/ksql/function/UdfLoaderUtil.java

@@ -40,6 +40,7 @@ public static UdfFactory createTestUdfFactory(final KsqlScalarFunction udf) {
         udf.getDescription(),
         "Test Author",
         "",
+        FunctionCategory.OTHER,
         KsqlScalarFunction.INTERNAL_PATH
     );
 

ksqldb-engine/src/main/java/io/confluent/ksql/function/UdtfLoader.java

@@ -74,6 +74,7 @@ public void loadUdtfFromClass(
         udtfDescriptionAnnotation.description(),
         udtfDescriptionAnnotation.author(),
         udtfDescriptionAnnotation.version(),
+        udtfDescriptionAnnotation.category(),
         path
     );
 

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArrayDistinct.java

@@ -15,6 +15,7 @@
 package io.confluent.ksql.function.udf.array;
 
 import com.google.common.collect.Sets;
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -24,6 +25,7 @@
 
 @UdfDescription(
     name = "array_distinct",
+    category = FunctionCategory.ARRAY,
     description = "Returns an array of all the distinct values, including NULL if present, from"
         + " the input array."
         + " The output array elements will be in order of their first occurrence in the input."

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArrayExcept.java

@@ -14,6 +14,7 @@
 
 package io.confluent.ksql.function.udf.array;
 
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -25,6 +26,7 @@
 
 @UdfDescription(
     name = "array_except",
+    category = FunctionCategory.ARRAY,
     description = "Returns an array of all the elements in an array except for those also present"
         + " in a second array. The order of entries in the first array is preserved although any"
         + " duplicates are removed. Returns NULL if either input is NULL.")

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArrayIntersect.java

@@ -16,6 +16,7 @@
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -24,6 +25,7 @@
 
 @UdfDescription(
     name = "array_intersect",
+    category = FunctionCategory.ARRAY,
     description = "Returns an array of all the distinct elements from the intersection of both"
         + " input arrays, or NULL if either input array is NULL. The order of entries in the"
         + " output is the same as in the first input array.")

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArrayJoin.java

@@ -16,6 +16,7 @@
 package io.confluent.ksql.function.udf.array;
 
 import com.google.common.collect.ImmutableSet;
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.KsqlFunctionException;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
@@ -29,6 +30,7 @@
 @SuppressWarnings("MethodMayBeStatic") // UDF methods can not be static.
 @UdfDescription(
     name = "ARRAY_JOIN",
+    category = FunctionCategory.ARRAY,
     description = "joins the array elements into a flat string representation",
     author = KsqlConstants.CONFLUENT_AUTHOR
 )

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArrayLength.java

@@ -15,6 +15,7 @@
 
 package io.confluent.ksql.function.udf.array;
 
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -27,6 +28,7 @@
 @SuppressWarnings("MethodMayBeStatic") // UDF methods can not be static.
 @UdfDescription(
     name = "ARRAY_LENGTH",
+    category = FunctionCategory.ARRAY,
     description = "Returns the length on an array",
     author = KsqlConstants.CONFLUENT_AUTHOR
 )

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArrayMax.java

@@ -14,6 +14,7 @@
 
 package io.confluent.ksql.function.udf.array;
 
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -24,6 +25,7 @@
  */
 @UdfDescription(
     name = "array_max",
+    category = FunctionCategory.ARRAY,
     description = "Return the maximum value from within an array of primitive values, according to"
         + " their natural sort order. If the array is NULL, or contains only NULLs, return NULL.")
 public class ArrayMax {

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArrayMin.java

@@ -14,6 +14,7 @@
 
 package io.confluent.ksql.function.udf.array;
 
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -24,6 +25,7 @@
  */
 @UdfDescription(
     name = "array_min",
+    category = FunctionCategory.ARRAY,
     description = "Return the minimum value from within an array of primitive values, according to"
         + " their natural sort order. If the array is NULL, or contains only NULLs, return NULL.")
 public class ArrayMin {

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/array/ArraySort.java

@@ -18,6 +18,7 @@
 import static java.util.Comparator.nullsLast;
 
 import com.google.common.collect.Lists;
+import io.confluent.ksql.function.FunctionCategory;
 import io.confluent.ksql.function.udf.Udf;
 import io.confluent.ksql.function.udf.UdfDescription;
 import io.confluent.ksql.function.udf.UdfParameter;
@@ -29,6 +30,7 @@
  */
 @UdfDescription(
     name = "array_sort",
+    category = FunctionCategory.ARRAY,
     description = "Sort an array of primitive values, according to their natural sort order. Any "
         + "NULLs in the array will be placed at the end.")
 public class ArraySort {