Enable bytes and string data types for max/min udaf

ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/max/MaxAggFunctionFactory.java

@@ -52,6 +52,8 @@ public KsqlAggregateFunction createAggregateFunction(
       case DATE:
       case TIME:
       case TIMESTAMP:
+      case STRING:
+      case BYTES:
         return new MaxKudaf(FUNCTION_NAME, initArgs.udafIndex(), argSchema);
       default:
         throw new KsqlException(

ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/min/MinAggFunctionFactory.java

@@ -51,6 +51,8 @@ public KsqlAggregateFunction createAggregateFunction(
       case DATE:
       case TIME:
       case TIMESTAMP:
+      case STRING:
+      case BYTES:
         return new MinKudaf(FUNCTION_NAME, initArgs.udafIndex(), argSchema);
       default:
         throw new KsqlException(

ksqldb-engine/src/test/java/io/confluent/ksql/function/udaf/max/BytesMaxKudafTest.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2022 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.udaf.max;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import io.confluent.ksql.GenericKey;
+import io.confluent.ksql.function.AggregateFunctionInitArguments;
+import io.confluent.ksql.function.KsqlAggregateFunction;
+import io.confluent.ksql.function.udf.string.FromBytes;
+import io.confluent.ksql.function.udf.string.ToBytes;
+import io.confluent.ksql.schema.ksql.SqlArgument;
+import io.confluent.ksql.schema.ksql.types.SqlTypes;
+import io.confluent.ksql.util.BytesUtils.Encoding;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import org.apache.kafka.streams.kstream.Merger;
+import org.junit.Before;
+import org.junit.Test;
+
+public class BytesMaxKudafTest {
+
+  private ToBytes toBytesUDF;
+  private FromBytes fromBytesUDF;
+
+  @Before
+  public void setUp() {
+    toBytesUDF = new ToBytes();
+    fromBytesUDF = new FromBytes();
+  }
+
+  @Test
+  public void shouldFindCorrectMax() {
+    final MaxKudaf<ByteBuffer> bytesMaxKudaf = getMaxComparableKudaf();
+    final String[] values = new String[]{"C", "F", "B", "E", "A", "D", "B"};
+    String currentMax = "A";
+    for (final String val : values) {
+      currentMax = fromBytesUDF.fromBytes(
+          bytesMaxKudaf.aggregate(
+              toBytesUDF.toBytes(val, Encoding.ASCII.toString()),
+              toBytesUDF.toBytes(currentMax, Encoding.ASCII.toString())),
+          Encoding.ASCII.toString());
+    }
+    assertThat("F", equalTo(currentMax));
+  }
+
+  @Test
+  public void shouldHandleNull() {
+    final MaxKudaf<ByteBuffer> bytesMaxKudaf = getMaxComparableKudaf();
+    final String[] values = new String[]{"C", "F", "B", "E", "A", "D", "B"};
+    String currentMax = null;
+
+    // null before any aggregation
+    currentMax = fromBytesUDF.fromBytes(
+        bytesMaxKudaf.aggregate(
+            null,
+            toBytesUDF.toBytes(currentMax, Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(null, equalTo(currentMax));
+
+    // now send each value to aggregation and verify
+    for (final String val : values) {
+      currentMax = fromBytesUDF.fromBytes(
+          bytesMaxKudaf.aggregate(
+              toBytesUDF.toBytes(val, Encoding.ASCII.toString()),
+              toBytesUDF.toBytes(currentMax, Encoding.ASCII.toString())),
+          Encoding.ASCII.toString());
+    }
+    assertThat("F", equalTo(currentMax));
+
+    // null should not impact result
+    currentMax = fromBytesUDF.fromBytes(
+        bytesMaxKudaf.aggregate(
+            null,
+            toBytesUDF.toBytes(currentMax, Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat("F", equalTo(currentMax));
+  }
+
+  @Test
+  public void shouldFindCorrectMaxForMerge() {
+    final MaxKudaf bytesMaxKudaf = getMaxComparableKudaf();
+    final Merger<GenericKey, ByteBuffer> merger = bytesMaxKudaf.getMerger();
+    final String mergeResult1 = fromBytesUDF.fromBytes(
+        merger.apply(
+            null,
+            toBytesUDF.toBytes("B", Encoding.ASCII.toString()),
+            toBytesUDF.toBytes("D", Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(mergeResult1, equalTo("D"));
+    final String mergeResult2 = fromBytesUDF.fromBytes(
+        merger.apply(
+            null,
+            toBytesUDF.toBytes("P", Encoding.ASCII.toString()),
+            toBytesUDF.toBytes("F", Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(mergeResult2, equalTo("P"));
+    final String mergeResult3 = fromBytesUDF.fromBytes(
+        merger.apply(
+            null,
+            toBytesUDF.toBytes("A", Encoding.ASCII.toString()),
+            toBytesUDF.toBytes("K", Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(mergeResult3, equalTo("K"));
+  }
+
+  private MaxKudaf getMaxComparableKudaf() {
+    final KsqlAggregateFunction aggregateFunction = new MaxAggFunctionFactory()
+        .createAggregateFunction(Collections.singletonList(SqlArgument.of(SqlTypes.BYTES)),
+            AggregateFunctionInitArguments.EMPTY_ARGS);
+    assertThat(aggregateFunction, instanceOf(MaxKudaf.class));
+    return  (MaxKudaf) aggregateFunction;
+  }
+
+}

ksqldb-engine/src/test/java/io/confluent/ksql/function/udaf/max/StringMaxKudafTest.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2022 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.udaf.max;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import io.confluent.ksql.GenericKey;
+import io.confluent.ksql.function.AggregateFunctionInitArguments;
+import io.confluent.ksql.function.KsqlAggregateFunction;
+import io.confluent.ksql.schema.ksql.SqlArgument;
+import io.confluent.ksql.schema.ksql.types.SqlTypes;
+import java.util.Collections;
+import org.apache.kafka.streams.kstream.Merger;
+import org.junit.Test;
+
+public class StringMaxKudafTest {
+
+  @Test
+  public void shouldFindCorrectMax() {
+    final MaxKudaf<String> stringMaxKudaf = getMaxComparableKudaf();
+    final String[] values = new String[]{"C", "F", "B", "E", "A", "D", "B"};
+    String currentMax = "A";
+    for (final String val : values) {
+      currentMax = stringMaxKudaf.aggregate(val, currentMax);
+    }
+    assertThat("F", equalTo(currentMax));
+  }
+
+  @Test
+  public void shouldHandleNull() {
+    final MaxKudaf<String> stringMaxKudaf = getMaxComparableKudaf();
+    final String[] values = new String[]{"C", "F", "B", "E", "A", "D", "B"};
+    String currentMax = null;
+
+    // null before any aggregation
+    currentMax = stringMaxKudaf.aggregate(null, currentMax);
+    assertThat(null, equalTo(currentMax));
+
+    // now send each value to aggregation and verify
+    for (final String val : values) {
+      currentMax = stringMaxKudaf.aggregate(val, currentMax);
+    }
+    assertThat("F", equalTo(currentMax));
+
+    // null should not impact result
+    currentMax = stringMaxKudaf.aggregate(null, currentMax);
+    assertThat("F", equalTo(currentMax));
+  }
+
+  @Test
+  public void shouldFindCorrectMaxForMerge() {
+    final MaxKudaf stringMaxKudaf = getMaxComparableKudaf();
+    final Merger<GenericKey, String> merger = stringMaxKudaf.getMerger();
+    final String mergeResult1 = merger.apply(null, "B", "D");
+    assertThat(mergeResult1, equalTo("D"));
+    final String mergeResult2 = merger.apply(null, "P", "F");
+    assertThat(mergeResult2, equalTo("P"));
+    final String mergeResult3 = merger.apply(null, "A", "K");
+    assertThat(mergeResult3, equalTo("K"));
+  }
+
+  private MaxKudaf getMaxComparableKudaf() {
+    final KsqlAggregateFunction aggregateFunction = new MaxAggFunctionFactory()
+        .createAggregateFunction(Collections.singletonList(SqlArgument.of(SqlTypes.STRING)),
+            AggregateFunctionInitArguments.EMPTY_ARGS);
+    assertThat(aggregateFunction, instanceOf(MaxKudaf.class));
+    return  (MaxKudaf) aggregateFunction;
+  }
+
+}

ksqldb-engine/src/test/java/io/confluent/ksql/function/udaf/min/BytesMinKudafTest.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2022 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.udaf.min;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import io.confluent.ksql.GenericKey;
+import io.confluent.ksql.function.AggregateFunctionInitArguments;
+import io.confluent.ksql.function.KsqlAggregateFunction;
+import io.confluent.ksql.function.udf.string.FromBytes;
+import io.confluent.ksql.function.udf.string.ToBytes;
+import io.confluent.ksql.schema.ksql.SqlArgument;
+import io.confluent.ksql.schema.ksql.types.SqlTypes;
+import io.confluent.ksql.util.BytesUtils.Encoding;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import org.apache.kafka.streams.kstream.Merger;
+import org.junit.Before;
+import org.junit.Test;
+
+public class BytesMinKudafTest {
+
+  private ToBytes toBytesUDF;
+  private FromBytes fromBytesUDF;
+
+  @Before
+  public void setUp() {
+    toBytesUDF = new ToBytes();
+    fromBytesUDF = new FromBytes();
+  }
+
+  @Test
+  public void shouldFindCorrectMin() {
+    final MinKudaf<ByteBuffer> bytesMinKudaf = getBytesMinKudaf();
+    final String[] values = new String[]{"C", "F", "B", "E", "A", "D", "B"};
+    String currentMin = "D";
+    for (final String val : values) {
+      currentMin = fromBytesUDF.fromBytes(
+          bytesMinKudaf.aggregate(
+              toBytesUDF.toBytes(val, Encoding.ASCII.toString()),
+              toBytesUDF.toBytes(currentMin, Encoding.ASCII.toString())),
+          Encoding.ASCII.toString());
+    }
+    assertThat("A", equalTo(currentMin));
+  }
+
+  @Test
+  public void shouldHandleNull() {
+    final MinKudaf<ByteBuffer> bytesMinKudaf = getBytesMinKudaf();
+    final String[] values = new String[]{"C", "F", "B", "E", "A", "D", "B"};
+    String currentMin = null;
+
+    // null before any aggregation
+    currentMin = fromBytesUDF.fromBytes(
+        bytesMinKudaf.aggregate(
+            null,
+            toBytesUDF.toBytes(currentMin, Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(null, equalTo(currentMin));
+
+    // now send each value to aggregation and verify
+    for (final String val : values) {
+      currentMin = fromBytesUDF.fromBytes(
+          bytesMinKudaf.aggregate(
+              toBytesUDF.toBytes(val, Encoding.ASCII.toString()),
+              toBytesUDF.toBytes(currentMin, Encoding.ASCII.toString())),
+          Encoding.ASCII.toString());
+    }
+    assertThat("A", equalTo(currentMin));
+
+    // null should not impact result
+    currentMin = fromBytesUDF.fromBytes(
+        bytesMinKudaf.aggregate(
+            null,
+            toBytesUDF.toBytes(currentMin, Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat("A", equalTo(currentMin));
+  }
+
+  @Test
+  public void shouldFindCorrectMinForMerge() {
+    final MinKudaf bytesMinKudaf = getBytesMinKudaf();
+    final Merger<GenericKey, ByteBuffer> merger = bytesMinKudaf.getMerger();
+    final String mergeResult1 = fromBytesUDF.fromBytes(
+        merger.apply(
+            null,
+            toBytesUDF.toBytes("B", Encoding.ASCII.toString()),
+            toBytesUDF.toBytes("D", Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(mergeResult1, equalTo("B"));
+    final String mergeResult2 = fromBytesUDF.fromBytes(
+        merger.apply(
+            null,
+            toBytesUDF.toBytes("P", Encoding.ASCII.toString()),
+            toBytesUDF.toBytes("F", Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(mergeResult2, equalTo("F"));
+    final String mergeResult3 = fromBytesUDF.fromBytes(
+        merger.apply(
+            null,
+            toBytesUDF.toBytes("A", Encoding.ASCII.toString()),
+            toBytesUDF.toBytes("K", Encoding.ASCII.toString())),
+        Encoding.ASCII.toString());
+    assertThat(mergeResult3, equalTo("A"));
+  }
+
+  private MinKudaf getBytesMinKudaf() {
+    final KsqlAggregateFunction aggregateFunction = new MinAggFunctionFactory()
+        .createAggregateFunction(Collections.singletonList(SqlArgument.of(SqlTypes.BYTES)),
+            AggregateFunctionInitArguments.EMPTY_ARGS);
+    assertThat(aggregateFunction, instanceOf(MinKudaf.class));
+    return  (MinKudaf) aggregateFunction;
+  }
+
+}