Cherry-pick stragglers for ksqlDB 0.10.0 docs (DOCS-4744) (#5687)

* feat: implements ARRAY_JOIN as requested in (#5028) (#5474) (#5638)

Co-authored-by: Hans-Peter Grahsl <hpgrahsl@users.noreply.github.com>

* feat: new split_to_map udf (#5563)

New UDF split_to_map(input, entryDelimiter, kvDelimiter) to build a map from a string.

Useful for taking messages from upstream systems and converting them into a more structured and usable format.

* feat: add CHR UDF (#5559)

A new UDF, CHR, to turn a number representing a unicode codepoint into a single-character string. Very useful for dealing with non-printable characters (tab, CR, LF, ...) in strings or those characters not easily represented in your local codepage.

Co-authored-by: Steven Zhang <35498506+stevenpyzhang@users.noreply.github.com>
Co-authored-by: Hans-Peter Grahsl <hpgrahsl@users.noreply.github.com>
Co-authored-by: Nick Dearden <blueedgenick@users.noreply.github.com>

docs/developer-guide/ksqldb-reference/scalar-functions.md

@@ -400,8 +400,39 @@ SLICE(col1, from, to)
 Slices a list based on the supplied indices. The indices start at 1 and
 include both endpoints.
 
+### `ARRAY_JOIN`
+
+```sql
+ARRAY_JOIN(col1, delimiter)
+```
+
+Creates a flat string representation of all the elements contained in the given array.
+The elements in the resulting string are separated by the chosen `delimiter`, 
+which is an optional parameter that falls back to a comma `,`. The current implementation only
+allows for array elements of primitive ksqlDB types.
+
 ## Strings
 
+### `CHR`
+
+```sql
+CHR(decimal_code | utf_string)
+```
+
+Returns a single-character string representing the Unicode code-point described by the input. The input parameter can be either a decimal character code or a string representation of a UTF code.
+
+Returns NULL if the input is NULL or does not represent a valid code-point.
+
+Commonly used to insert control characters such as `Tab` (9), `Line Feed` (10), or `Carriage Return` (13) into strings.
+
+Examples:
+```sql
+CHR(75)        => 'K'
+CHR('\u004b')  => 'K'
+CHR(22909)     => '好'
+CHR('\u597d')  => '好'
+```
+
 ### `CONCAT`
 
 ```sql
@@ -721,6 +752,23 @@ If the delimiter is found at the beginning or end
 of the string, or there are contiguous delimiters,
 then an empty space is added to the array.
 
+### `SPLIT_TO_MAP`
+
+```sql
+SPLIT_TO_MAP(input, entryDelimiter, kvDelimiter)
+```
+
+Splits a string into key-value pairs and creates a map from them. The 
+`entryDelimiter` splits the string into key-value pairs which are then split by `kvDelimiter`. If the same key is present multiple times in the input, the latest value for that key is returned. 
+
+Returns NULL if the input text is NULL.
+Returns NULL if either of the delimiters is NULL or an empty string.
+
+Example:
+```sql
+SPLIT_TO_MAP('apple':='green'/'cherry':='red', '/', ':=')  => { 'apple':'green', 'cherry':'red'}
+```
+
 ### `SUBSTRING`
 
 ```sql

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

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2020 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
+ *
+ * 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.
+ */
+
+package io.confluent.ksql.function.udf.array;
+
+import com.google.common.collect.ImmutableSet;
+import io.confluent.ksql.function.KsqlFunctionException;
+import io.confluent.ksql.function.udf.Udf;
+import io.confluent.ksql.function.udf.UdfDescription;
+import io.confluent.ksql.function.udf.UdfParameter;
+import io.confluent.ksql.util.KsqlConstants;
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Set;
+import java.util.StringJoiner;
+
+@SuppressWarnings("MethodMayBeStatic") // UDF methods can not be static.
+@UdfDescription(
+    name = "ARRAY_JOIN",
+    description = "joins the array elements into a flat string representation",
+    author = KsqlConstants.CONFLUENT_AUTHOR
+)
+public class ArrayJoin {
+
+  private static final String DEFAULT_DELIMITER = ",";
+  private static final Set<Class> KSQL_PRIMITIVES = ImmutableSet.of(
+      Boolean.class,Integer.class,Long.class,Double.class,BigDecimal.class,String.class
+  );
+
+  @Udf
+  public <T> String join(
+      @UdfParameter(description = "the array to join using the default delimiter '"
+          + DEFAULT_DELIMITER + "'") final List<T> array
+  ) {
+    return join(array, DEFAULT_DELIMITER);
+  }
+
+  @Udf
+  public <T> String join(
+      @UdfParameter(description = "the array to join using the specified delimiter")
+      final List<T> array,
+      @UdfParameter(description = "the string to be used as element delimiter")
+      final String delimiter
+  ) {
+
+    if (array == null) {
+      return null;
+    }
+
+    final StringJoiner sj = new StringJoiner(delimiter == null ? "" : delimiter);
+    array.forEach(e -> processElement(e, sj));
+    return sj.toString();
+
+  }
+
+  @SuppressWarnings("unchecked")
+  private static <T> void processElement(final T element, final StringJoiner joiner) {
+
+    if (element == null || KSQL_PRIMITIVES.contains(element.getClass())) {
+      handlePrimitiveType(element, joiner);
+    } else {
+      throw new KsqlFunctionException("error: hit element of type "
+          + element.getClass().getTypeName() + " which is currently not supported");
+    }
+
+  }
+
+  private static void handlePrimitiveType(final Object element, final StringJoiner joiner) {
+    joiner.add(element != null ? element.toString() : null);
+  }
+
+}

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/string/Chr.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2020 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
+ *
+ * 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.
+ */
+
+package io.confluent.ksql.function.udf.string;
+
+import io.confluent.ksql.function.udf.Udf;
+import io.confluent.ksql.function.udf.UdfDescription;
+import io.confluent.ksql.function.udf.UdfParameter;
+import org.apache.commons.lang3.StringEscapeUtils;
+
+@UdfDescription(
+    name = "Chr",
+    description = "Returns a single-character string corresponding to the input character code.")
+public class Chr {
+
+  @Udf
+  public String chr(@UdfParameter(
+      description = "Decimal codepoint") final Integer decimalCode) {
+    if (decimalCode == null) {
+      return null;
+    }
+    if (!Character.isValidCodePoint(decimalCode)) {
+      return null;
+    }
+    final char[] resultChars = Character.toChars(decimalCode.intValue());
+    return String.valueOf(resultChars);
+  }
+
+  @Udf
+  public String chr(@UdfParameter(
+      description = "UTF16 code for the desired character e.g. '\\u004b'") final String utf16Code) {
+    if (utf16Code == null || utf16Code.length() < 6 || !utf16Code.startsWith("\\u")) {
+      return null;
+    }
+    return StringEscapeUtils.unescapeJava(utf16Code);
+  }
+}

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/string/SplitToMap.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2020 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
+ *
+ * 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.
+ */
+
+package io.confluent.ksql.function.udf.string;
+
+import com.google.common.base.Splitter;
+import io.confluent.ksql.function.udf.Udf;
+import io.confluent.ksql.function.udf.UdfDescription;
+import io.confluent.ksql.function.udf.UdfParameter;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+@UdfDescription(
+    name = "split_to_map",
+    description = "Splits a string into key-value pairs and creates a map from them. The "
+        + "'entryDelimiter' splits the string into key-value pairs which are then split by "
+        + "'kvDelimiter'. If the same key is present multiple times in the input, the latest "
+        + "value for that key is returned. Returns NULL f the input text or either of the "
+        + "delimiters is NULL.")
+public class SplitToMap {
+  @Udf
+  public Map<String, String> splitToMap(
+      @UdfParameter(
+          description = "Separator string and values to join") final String input,
+      @UdfParameter(
+          description = "Separator string and values to join") final String entryDelimiter,
+      @UdfParameter(
+          description = "Separator string and values to join") final String kvDelimiter) {
+
+    if (input == null || entryDelimiter == null || kvDelimiter == null) {
+      return null;
+    }
+
+    if (entryDelimiter.isEmpty() || kvDelimiter.isEmpty() || entryDelimiter.equals(kvDelimiter)) {
+      return null;
+    }
+
+    final Iterable<String> entries = Splitter.on(entryDelimiter).omitEmptyStrings().split(input);
+    final Map<String, String> output = StreamSupport.stream(entries.spliterator(), false)
+        .filter(e -> e.contains(kvDelimiter))
+        .map(kv -> Splitter.on(kvDelimiter).split(kv).iterator())
+        .collect(Collectors.toMap(
+            kvIter -> kvIter.next(), 
+            kvIter -> kvIter.next(),
+            (v1, v2) -> v2));
+
+    return output;
+  }
+}

ksqldb-engine/src/test/java/io/confluent/ksql/function/udf/array/ArrayJoinTest.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2020 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
+ *
+ * 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.
+ */
+
+package io.confluent.ksql.function.udf.array;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThrows;
+
+import io.confluent.ksql.function.KsqlFunctionException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import org.junit.Test;
+
+public class ArrayJoinTest {
+
+  private static final String CUSTOM_DELIMITER = "|";
+
+  private final ArrayJoin arrayJoinUDF = new ArrayJoin();
+
+  @Test
+  public void shouldReturnNullForNullInput() {
+    assertThat(arrayJoinUDF.join(null), nullValue());
+    assertThat(arrayJoinUDF.join(null,CUSTOM_DELIMITER), nullValue());
+  }
+
+  @Test
+  public void shouldReturnEmptyStringForEmptyArrays() {
+    assertThat(arrayJoinUDF.join(Collections.emptyList()).isEmpty(),is(true));
+    assertThat(arrayJoinUDF.join(Collections.emptyList(),CUSTOM_DELIMITER).isEmpty(),is(true));
+  }
+
+  @Test
+  public void shouldReturnCorrectStringForFlatArraysWithPrimitiveTypes() {
+
+    assertThat(arrayJoinUDF.join(Arrays.asList(true, null, false),""),
+        is("truenullfalse")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList(true, null, false)),
+        is("true,null,false")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList(true,null,false),CUSTOM_DELIMITER),
+        is("true"+CUSTOM_DELIMITER+"null"+CUSTOM_DELIMITER+"false")
+    );
+
+    assertThat(arrayJoinUDF.join(Arrays.asList(1,23,-42,0),null), is("123-420"));
+    assertThat(arrayJoinUDF.join(Arrays.asList(1,23,-42,0)), is("1,23,-42,0"));
+    assertThat(arrayJoinUDF.join(Arrays.asList(1,23,-42,0),CUSTOM_DELIMITER),
+        is("1"+CUSTOM_DELIMITER+"23"+CUSTOM_DELIMITER+"-42"+CUSTOM_DELIMITER+"0")
+    );
+
+    assertThat(arrayJoinUDF.join(Arrays.asList(-4294967297L, 8589934592L),""),
+        is("-42949672978589934592")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList(-4294967297L, 8589934592L)),
+        is("-4294967297,8589934592")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList(-4294967297L, 8589934592L), CUSTOM_DELIMITER),
+        is("-4294967297"+CUSTOM_DELIMITER+"8589934592")
+    );
+
+    assertThat(arrayJoinUDF.join(Arrays.asList(1.23,-23.42,0.0),null),
+        is("1.23-23.420.0")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList(1.23,-23.42,0.0)),
+        is("1.23,-23.42,0.0")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList(1.23,-23.42,0.0),CUSTOM_DELIMITER),
+        is("1.23"+CUSTOM_DELIMITER+"-23.42"+CUSTOM_DELIMITER+"0.0")
+    );
+
+    assertThat(arrayJoinUDF.join(
+        Arrays.asList(new BigDecimal("123.45"), new BigDecimal("987.65")),null
+        ),
+        is("123.45987.65")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList(new BigDecimal("123.45"), new BigDecimal("987.65"))),
+        is("123.45,987.65")
+    );
+    assertThat(arrayJoinUDF.join(
+        Arrays.asList(new BigDecimal("123.45"), new BigDecimal("987.65")),CUSTOM_DELIMITER),
+        is("123.45"+CUSTOM_DELIMITER+"987.65")
+    );
+
+    assertThat(arrayJoinUDF.join(Arrays.asList("Hello","From","","Ksqldb","Udf"),""),
+        is("HelloFromKsqldbUdf")
+    );
+    assertThat(arrayJoinUDF.join(Arrays.asList("Hello","From","","Ksqldb","Udf")),
+        is("Hello,From,,Ksqldb,Udf")
+    );
+    assertThat(
+        arrayJoinUDF.join(Arrays.asList("hello","from","","ksqldb","udf",null),CUSTOM_DELIMITER),
+        is("hello"+CUSTOM_DELIMITER+"from"+CUSTOM_DELIMITER+CUSTOM_DELIMITER
+            +"ksqldb"+CUSTOM_DELIMITER+"udf"+CUSTOM_DELIMITER+"null")
+    );
+
+  }
+
+  @Test
+  public void shouldThrowExceptionForExamplesOfUnsupportedElementTypes() {
+    assertThrows(KsqlFunctionException.class,
+        () -> arrayJoinUDF.join(Arrays.asList('a','b')));
+    assertThrows(KsqlFunctionException.class,
+        () -> arrayJoinUDF.join(Arrays.asList(BigInteger.ONE,BigInteger.ZERO)));
+    assertThrows(KsqlFunctionException.class,
+        () -> arrayJoinUDF.join(Arrays.asList(-23.0f,42.42f,0.0f)));
+    assertThrows(KsqlFunctionException.class,
+        () -> arrayJoinUDF.join(Arrays.asList(
+            new HashSet<>(Arrays.asList("foo", "blah")),
+            new HashSet<>(Arrays.asList("ksqlDB", "UDF"))
+        ))
+    );
+  }
+
+}