[FLINK-6924] [table] Add Table API log() function.

This closes apache#5638.

docs/dev/table/sql.md

@@ -1361,6 +1361,19 @@ LOG10(numeric)
  </td>
  </tr>
 
+ <tr>
+  <td>
+  {% highlight text %}
+LOG(x numeric)
+LOG(b numeric, x numeric)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Returns the logarithm of a <i>numeric</i>.</p>
+ <p>If called with one parameter, this function returns the natural logarithm of <code>x</code>. If called with two parameters, this function returns the logarithm of <code>x</code> to the base <code>b</code>. <code>x</code> must be greater than 0. <code>b</code> must be greater than 1.</p>
+ </td>
+ </tr>
+
  <tr>
   <td>
   {% highlight text %}
@@ -1581,17 +1594,6 @@ RAND_INTEGER(seed integer, bound integer)
  </td>
  </tr>
 
- <tr>
-  <td>
-  {% highlight text %}
-LOG(x numeric), LOG(base numeric, x numeric)
-{% endhighlight %}
- </td>
- <td>
- <p>Returns the natural logarithm of a specified number of a specified base. If called with one parameter, this function returns the natural logarithm of <code>x</code>. If called with two parameters, this function returns the logarithm of <code>x</code> to the base <code>b</code>. <code>x</code> must be greater than 0. <code>b</code> must be greater than 1.</p>
- </td>
- </tr>
-
  <tr>
   <td>
 {% highlight text %}

docs/dev/table/tableApi.md

@@ -2030,6 +2030,19 @@ NUMERIC.log10()
  <p>Calculates the base 10 logarithm of given value.</p>
  </td>
  </tr>
+
+ <tr>
+ <td>
+ {% highlight java %}
+numeric1.log()
+numeric1.log(numeric2)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Calculates the logarithm of a given numeric value.</p>
+ <p>If called without a parameter, this function returns the natural logarithm of <code>numeric1</code>. If called with a parameter <code>numeric2</code>, this function returns the logarithm of <code>numeric1</code> to the base <code>numeric2</code>. <code>numeric1</code> must be greater than 0. <code>numeric2</code> must be greater than 1.</p>
+ </td>
+ </tr>
 
  <tr>
   <td>
@@ -3512,6 +3525,19 @@ NUMERIC.log10()
  </td>
  </tr>
 
+ <tr>
+  <td>
+  {% highlight scala %}
+numeric1.log()
+numeric1.log(numeric2)
+{% endhighlight %}
+ </td>
+ <td>
+ <p>Calculates the logarithm of a given numeric value.</p>
+ <p>If called without a parameter, this function returns the natural logarithm of <code>numeric1</code>. If called with a parameter <code>numeric2</code>, this function returns the logarithm of <code>numeric1</code> to the base <code>numeric2</code>. <code>numeric1</code> must be greater than 0. <code>numeric2</code> must be greater than 1.</p>
+ </td>
+ </tr>
+
  <tr>
   <td>
   {% highlight scala %}

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/expressionDsl.scala

@@ -296,15 +296,25 @@ trait ImplicitExpressionOperations {
  def exp() = Exp(expr)
 
  /**
- * Calculates the base 10 logarithm of given value.
+ * Calculates the base 10 logarithm of the given value.
  */
  def log10() = Log10(expr)
 
  /**
- * Calculates the natural logarithm of given value.
+ * Calculates the natural logarithm of the given value.
  */
  def ln() = Ln(expr)
 
+ /**
+ * Calculates the natural logarithm of the given value.
+ */
+ def log() = Log(null, expr)
+
+ /**
+ * Calculates the logarithm of the given value to the given base.
+ */
+ def log(base: Expression) = Log(base, expr)
+
  /**
  * Calculates the given number raised to the power of the other value.
  */

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/ExpressionParser.scala

@@ -57,6 +57,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
  lazy val EXTRACT: Keyword = Keyword("extract")
  lazy val FLOOR: Keyword = Keyword("floor")
  lazy val CEIL: Keyword = Keyword("ceil")
+ lazy val LOG: Keyword = Keyword("log")
  lazy val YEARS: Keyword = Keyword("years")
  lazy val YEAR: Keyword = Keyword("year")
  lazy val MONTHS: Keyword = Keyword("months")
@@ -246,6 +247,12 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
  case operand ~ _ ~ _ ~ _ ~ unit ~ _ => TemporalCeil(unit, operand)
  }
 
+ // required because op.log(base) changes order of a parameters
+ lazy val suffixLog: PackratParser[Expression] =
+ composite ~ "." ~ LOG ~ "(" ~ expression ~ ")" ^^ {
+ case operand ~ _ ~ _ ~ _ ~ base ~ _ => Log(base, operand)
+ }
+
  lazy val suffixFunctionCall: PackratParser[Expression] =
  composite ~ "." ~ functionIdent ~ "(" ~ repsep(expression, ",") ~ ")" ^^ {
  case operand ~ _ ~ name ~ _ ~ args ~ _ => Call(name.toUpperCase, operand :: args)
@@ -307,6 +314,8 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
  // expressions that need special expression conversion
  suffixAs | suffixTimeInterval | suffixRowInterval | suffixToTimestamp | suffixToTime |
  suffixToDate |
+ // expression for log
+ suffixLog |
  // expressions that take enumerations
  suffixCast | suffixTrim | suffixTrimWithoutArgs | suffixExtract | suffixFloor | suffixCeil |
  // expressions that take literals

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/mathExpressions.scala

@@ -92,6 +92,28 @@ case class Log10(child: Expression) extends UnaryExpression with InputTypeSpec {
  }
 }
 
+case class Log(base: Expression, antilogarithm: Expression) extends Expression with InputTypeSpec {
+ def this(antilogarithm: Expression) = this(null, antilogarithm)
+
+ override private[flink] def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
+
+ override private[flink] def children: Seq[Expression] =
+ if (base == null) Seq(antilogarithm) else Seq(base, antilogarithm)
+
+ override private[flink] def expectedTypes: Seq[TypeInformation[_]] =
+ Seq.fill(children.length)(DOUBLE_TYPE_INFO)
+
+ override def toString: String = s"log(${children.mkString(",")})"
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
+ relBuilder.call(ScalarSqlFunctions.LOG, children.map(_.toRexNode))
+ }
+}
+
+object Log {
+ def apply(antilogarithm: Expression): Log = Log(null, antilogarithm)
+}
+
 case class Ln(child: Expression) extends UnaryExpression with InputTypeSpec {
  override private[flink] def resultType: TypeInformation[_] = DOUBLE_TYPE_INFO
 

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala

@@ -212,6 +212,7 @@ object FunctionCatalog {
  "floor" -> classOf[Floor],
  "log10" -> classOf[Log10],
  "ln" -> classOf[Ln],
+ "log" -> classOf[Log],
  "power" -> classOf[Power],
  "mod" -> classOf[Mod],
  "sqrt" -> classOf[Sqrt],

flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala

@@ -1286,30 +1286,52 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
 
  @Test
  def testLog(): Unit = {
- testSqlApi(
+ testAllApis(
+ 'f6.log(),
+ "f6.log",
  "LOG(f6)",
  "1.5260563034950492"
  )
 
- testSqlApi(
- "LOG(f6-f6 + 10, f6-f6+100)",
+ testTableApi(
+ Log('f6),
+ "Log(f6)",
+ "1.5260563034950492"
+ )
+
+ testAllApis(
+ ('f6 - 'f6 + 100).log('f6 - 'f6 + 10),
+ "(f6 - f6 + 100).log(f6 - f6 + 10)",
+ "LOG(f6 - f6 + 10, f6 - f6 + 100)",
  "2.0"
  )
 
- testSqlApi(
+ testAllApis(
+ ('f6 + 20).log(),
+ "(f6+20).log",
  "LOG(f6+20)",
  "3.202746442938317"
  )
 
- testSqlApi(
+ testAllApis(
+ 10.log(),
+ "10.log",
  "LOG(10)",
  "2.302585092994046"
  )
 
- testSqlApi(
+ testAllApis(
+ 100.log(10),
+ "100.log(10)",
  "LOG(10, 100)",
  "2.0"
  )
+
+ testTableApi(
+ Log(10, 100),
+ "Log(10, 100)",
+ "2.0"
+ )
  }
 
  // ----------------------------------------------------------------------------------------------