@Udf(description = "Splits a string into an array of substrings based on a delimiter.")
  public List<String> split(
      @UdfParameter(
          description = "The string to be split. If NULL, then function returns NULL.")
      final String string,
      @UdfParameter(
          description = "The delimiter to split a string by. If NULL, then function returns NULL.")
      final String delimiter) {
    if (string == null || delimiter == null) {
      return null;
    }

    // Java split() accepts regular expressions as a delimiter, but the behavior of this UDF split()
    // is to accept only literal strings. This method uses Guava Splitter instead, which does not
    // accept any regex pattern. This is to avoid a confusion to users when splitting by regex
    // special characters, such as '.' and '|'.

    try {
      // Guava Splitter does not accept empty delimiters. Use the Java split() method instead.
      if (delimiter.isEmpty()) {
        return Arrays.asList(EMPTY_DELIMITER.split(string));
      } else {
        return Splitter.on(delimiter).splitToList(string);
      }
    } catch (final Exception e) {
      throw new KsqlFunctionException(
          String.format("Invalid delimiter '%s' in the split() function.", delimiter), e);
    }
  }
@Test
  public void shouldSplitAndAddEmptySpacesIfDelimiterBytesIsFoundAtTheBeginningOrEnd() {
    final ByteBuffer aBytes = ByteBuffer.wrap(new byte[]{'A'});
    final ByteBuffer bBytes = ByteBuffer.wrap(new byte[]{'B'});
    final ByteBuffer dollarBytes = ByteBuffer.wrap(new byte[]{'$'});

    assertThat(
        splitUdf.split(
            ByteBuffer.wrap(new byte[]{'$','A'}),
            dollarBytes),
        contains(EMPTY_BYTES, aBytes));

    assertThat(
        splitUdf.split(
            ByteBuffer.wrap(new byte[]{'$','A','$','B'}),
            dollarBytes),
        contains(EMPTY_BYTES, aBytes, bBytes));

    assertThat(
        splitUdf.split(
            ByteBuffer.wrap(new byte[]{'A','$'}),
            dollarBytes),
        contains(aBytes, EMPTY_BYTES));

    assertThat(
        splitUdf.split(
            ByteBuffer.wrap(new byte[]{'A','$','B','$'}),
            dollarBytes),
        contains(aBytes, bBytes, EMPTY_BYTES));

    assertThat(
        splitUdf.split(
            ByteBuffer.wrap(new byte[]{'$','A','$','B','$'}),
            dollarBytes),
        contains(EMPTY_BYTES, aBytes, bBytes, EMPTY_BYTES));
  }