feat: Add log, power, and cbrt UDFs (#9366)

* test: Add unit tests for log, power, cbrt

* feat: add cbrt, log, and power functions

* test: Add QTT tests for log, power, and cbrt

* test: Add historical plans for log, power, and cbrt

* docs: Add docs for CBRT, POWER, LOG

* test: use closeTo matching for floating-point vals

* docs: bump ksql version for new functions

* fix: incorrect desc for cbrt overloads

* Apply suggestions from code review

Co-authored-by: Jim Galasyn <jim.galasyn@confluent.io>

Co-authored-by: Jim Galasyn <jim.galasyn@confluent.io>

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

@@ -116,6 +116,16 @@ Converts one type to another. The following casts are supported:
 
 ---
 
+### **`CBRT`**
+
+```sql title="Since: 0.29.0"
+CBRT(col1)
+```
+
+Returns the cube root of `col1`.
+
+---
+
 ### **`CEIL`**
 
 ```sql title="Since: 0.1.0"
@@ -268,6 +278,19 @@ The value of `col1` must be greater than 0.
 
 ---
 
+### **`LOG`**
+
+```sql title="Since: 0.29.0"
+LOG(value)
+LOG(base, value)
+```
+
+The single-parameter version of this method returns the base 10 logarithm of the `value`. The two-parameter version returns the logarithm with the given `base` of the `value`.
+
+This function returns `-Infinity` for any `base` when the `value` is `0`. It returns `NaN` when the `value` is negative, when the `base` is negative, when the `base` is `0`, or when the `base` is `1`.
+
+---
+
 ### **`PI`**
 
 ```sql title="Since: 0.28.0"
@@ -278,6 +301,18 @@ Returns an approximate value of _π_.
 
 ---
 
+### **`POWER`**
+
+```sql title="Since: 0.29.0"
+POWER(base, exponent)
+```
+
+Calculates the value of the `base` raised to the `exponent`.
+
+This function returns `Infinity` when the result overflows the `DOUBLE` type.
+
+---
+
 ### **`RADIANS`**
 
 ```sql title="Since: 0.28.0"

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/math/Cbrt.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2022 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.math;
+
+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;
+import io.confluent.ksql.util.KsqlConstants;
+
+@UdfDescription(
+        name = "cbrt",
+        category = FunctionCategory.MATHEMATICAL,
+        author = KsqlConstants.CONFLUENT_AUTHOR,
+        description = "The cube root of a value."
+)
+public class Cbrt {
+
+  @Udf(description = "Returns the cube root of an INT value")
+  public Double cbrt(
+          @UdfParameter(
+                  value = "value",
+                  description = "The value to get the cube root of."
+          ) final Integer value
+  ) {
+    return cbrt(value == null ? null : value.doubleValue());
+  }
+
+  @Udf(description = "Returns the cube root of a BIGINT value")
+  public Double cbrt(
+          @UdfParameter(
+                  value = "value",
+                  description = "The value to get the cube root of."
+          ) final Long value
+  ) {
+    return cbrt(value == null ? null : value.doubleValue());
+  }
+
+  @Udf(description = "Returns the cube root of a DOUBLE value")
+  public Double cbrt(
+          @UdfParameter(
+                  value = "value",
+                  description = "The value to get the cube root of."
+          ) final Double value
+  ) {
+    return value == null
+            ? null
+            : Math.cbrt(value);
+  }
+}

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/math/Log.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2022 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.math;
+
+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;
+import io.confluent.ksql.util.KsqlConstants;
+
+@UdfDescription(
+        name = "log",
+        category = FunctionCategory.MATHEMATICAL,
+        author = KsqlConstants.CONFLUENT_AUTHOR,
+        description = "The logarithm of a value."
+)
+public class Log {
+
+  @Udf(description = "Returns the base 10 logarithm of an INT value.")
+  public Double log(
+          @UdfParameter(
+                  value = "value",
+                  description = "the value get the base 10 logarithm of."
+          ) final Integer value
+  ) {
+    return log(value == null ? null : value.doubleValue());
+  }
+
+  @Udf(description = "Returns the base 10 logarithm of a BIGINT value.")
+  public Double log(
+          @UdfParameter(
+                  value = "value",
+                  description = "the value get the base 10 logarithm of."
+          ) final Long value
+  ) {
+    return log(value == null ? null : value.doubleValue());
+  }
+
+  @Udf(description = "Returns the base 10 logarithm of a DOUBLE value.")
+  public Double log(
+          @UdfParameter(
+                  value = "value",
+                  description = "the value get the base 10 logarithm of."
+          ) final Double value
+  ) {
+    return value == null
+            ? null
+            : Math.log(value);
+  }
+
+  @Udf(description = "Returns the logarithm with the given base of an INT value.")
+  public Double log(
+          @UdfParameter(
+                  value = "base",
+                  description = "the base of the logarithm."
+          ) final Integer base,
+          @UdfParameter(
+                  value = "value",
+                  description = "the value get the logarithm of."
+          ) final Integer value
+  ) {
+    return log(
+            base == null ? null : base.doubleValue(),
+            value == null ? null : value.doubleValue()
+    );
+  }
+
+  @Udf(description = "Returns the logarithm with the given base of a BIGINT value.")
+  public Double log(
+          @UdfParameter(
+                  value = "base",
+                  description = "the base of the logarithm."
+          ) final Long base,
+          @UdfParameter(
+                  value = "value",
+                  description = "the value get the logarithm of."
+          ) final Long value
+  ) {
+    return log(
+            base == null ? null : base.doubleValue(),
+            value == null ? null : value.doubleValue()
+    );
+  }
+
+  @Udf(description = "Returns the logarithm with the given base of a DOUBLE value.")
+  public Double log(
+          @UdfParameter(
+                  value = "base",
+                  description = "the base of the logarithm."
+          ) final Double base,
+          @UdfParameter(
+                  value = "value",
+                  description = "the value get the logarithm of."
+          ) final Double value
+  ) {
+
+    if (base == null || value == null) {
+      return null;
+    }
+
+    if (base <= 0 || base == 1) {
+      return Double.NaN;
+    }
+
+    return Math.log(value) / Math.log(base);
+  }
+}

ksqldb-engine/src/main/java/io/confluent/ksql/function/udf/math/Power.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2022 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.math;
+
+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;
+import io.confluent.ksql.util.KsqlConstants;
+
+@UdfDescription(
+        name = "power",
+        category = FunctionCategory.MATHEMATICAL,
+        author = KsqlConstants.CONFLUENT_AUTHOR,
+        description = "Given a base and an exponent, returns the result of the base raised to the "
+                + "exponent."
+)
+public class Power {
+
+  @Udf(description = "Returns the INT base raised to the INT exponent.")
+  public Double power(
+          @UdfParameter(
+                  value = "base",
+                  description = "the base of the power."
+          ) final Integer base,
+          @UdfParameter(
+                  value = "exponent",
+                  description = "the exponent of the power."
+          ) final Integer exponent
+  ) {
+    return power(
+            base == null ? null : base.doubleValue(),
+            exponent == null ? null : exponent.doubleValue()
+    );
+  }
+
+  @Udf(description = "Returns the BIGINT base raised to the BIGINT exponent.")
+  public Double power(
+          @UdfParameter(
+                  value = "base",
+                  description = "the base of the power."
+          ) final Long base,
+          @UdfParameter(
+                  value = "exponent",
+                  description = "the exponent of the power."
+          ) final Long exponent
+  ) {
+    return power(
+            base == null ? null : base.doubleValue(),
+            exponent == null ? null : exponent.doubleValue()
+    );
+  }
+
+  @Udf(description = "Returns the DOUBLE base raised to the DOUBLE exponent.")
+  public Double power(
+          @UdfParameter(
+                  value = "base",
+                  description = "the base of the power."
+          ) final Double base,
+          @UdfParameter(
+                  value = "exponent",
+                  description = "the exponent of the power."
+          ) final Double exponent
+  ) {
+    return base == null || exponent == null ? null : Math.pow(base, exponent);
+  }
+}

ksqldb-engine/src/test/java/io/confluent/ksql/function/udf/math/CbrtTest.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2022 Confluent Inc.
+ *
+ * Licensed under the Confluent Community 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.math;
+
+import org.junit.Before;
+import org.junit.Test;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.closeTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+
+public class CbrtTest {
+
+  private Cbrt udf;
+
+  @Before
+  public void setUp() {
+    udf = new Cbrt();
+  }
+
+  @Test
+  public void shouldHandleNull() {
+    assertThat(udf.cbrt((Integer)null), is(nullValue()));
+    assertThat(udf.cbrt((Long)null), is(nullValue()));
+    assertThat(udf.cbrt((Double)null), is(nullValue()));
+  }
+
+  @Test
+  public void shouldHandleNegative() {
+    assertThat(udf.cbrt(-8), closeTo(-2.0, 0.000000000000001));
+    assertThat(udf.cbrt(-3L), closeTo(-1.4422495703074083, 0.000000000000001));
+    assertThat(udf.cbrt(-1.0), closeTo(-1.0, 0.000000000000001));
+  }
+
+  @Test
+  public void shouldHandleZero() {
+    assertThat(udf.cbrt(0), closeTo(0.0, 0.000000000000001));
+    assertThat(udf.cbrt(0L), closeTo(0.0, 0.000000000000001));
+    assertThat(udf.cbrt(0.0), closeTo(0.0, 0.000000000000001));
+  }
+
+  @Test
+  public void shouldHandlePositive() {
+    assertThat(udf.cbrt(8), closeTo(2.0, 0.000000000000001));
+    assertThat(udf.cbrt(3L), closeTo(1.4422495703074083, 0.000000000000001));
+    assertThat(udf.cbrt(1.0), closeTo(1.0, 0.000000000000001));
+  }
+}