flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From twal...@apache.org
Subject flink git commit: [FLINK-6891] [table] Add LOG support in SQL
Date Tue, 18 Jul 2017 09:56:07 GMT
Repository: flink
Updated Branches:
  refs/heads/master d7f9f9e3e -> 61d87d7b6


[FLINK-6891] [table] Add LOG support in SQL

This closes #4122.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/61d87d7b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/61d87d7b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/61d87d7b

Branch: refs/heads/master
Commit: 61d87d7b642080a1b64f5c42cff46d142641bf20
Parents: d7f9f9e
Author: sunjincheng121 <sunjincheng121@gmail.com>
Authored: Wed Jun 14 17:13:55 2017 +0800
Committer: twalthr <twalthr@apache.org>
Committed: Tue Jul 18 11:55:24 2017 +0200

----------------------------------------------------------------------
 docs/dev/table/sql.md                           | 11 ++++++++
 .../table/codegen/calls/BuiltInMethods.scala    |  6 +++++
 .../table/codegen/calls/FunctionGenerator.scala | 21 ++++++++++++---
 .../functions/sql/ScalarSqlFunctions.scala      | 12 ++++++++-
 .../runtime/functions/ScalarFunctions.scala     | 27 +++++++++++++++++++
 .../flink/table/validate/FunctionCatalog.scala  |  3 ++-
 .../table/expressions/ScalarFunctionsTest.scala | 28 ++++++++++++++++++++
 .../ScalarFunctionsValidationTest.scala         | 22 +++++++++++++++
 8 files changed, 124 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/docs/dev/table/sql.md
----------------------------------------------------------------------
diff --git a/docs/dev/table/sql.md b/docs/dev/table/sql.md
index 85a758b..42d56ec 100644
--- a/docs/dev/table/sql.md
+++ b/docs/dev/table/sql.md
@@ -1473,6 +1473,17 @@ 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>
+
   </tbody>
 </table>
 

http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
index b7da141..671ae07 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
@@ -25,6 +25,12 @@ import org.apache.calcite.runtime.SqlFunctions
 import org.apache.flink.table.runtime.functions.ScalarFunctions
 
 object BuiltInMethods {
+
+  val LOG = Types.lookupMethod(classOf[ScalarFunctions], "log", classOf[Double])
+
+  val LOG_WITH_BASE =
+    Types.lookupMethod(classOf[ScalarFunctions], "log", classOf[Double], classOf[Double])
+
   val LOG10 = Types.lookupMethod(classOf[Math], "log10", classOf[Double])
 
   val EXP = Types.lookupMethod(classOf[Math], "exp", classOf[Double])

http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
index d071279..f05b9ba 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, SqlTimeTypeInfo, TypeInformation}
 import org.apache.flink.api.java.typeutils.GenericTypeInfo
 import org.apache.flink.table.functions.sql.DateTimeSqlFunction
+import org.apache.flink.table.functions.sql.ScalarSqlFunctions
 import org.apache.flink.table.functions.utils.{ScalarSqlFunction, TableSqlFunction}
 import org.apache.flink.table.functions.sql.ScalarSqlFunctions._
 
@@ -377,6 +378,11 @@ object FunctionGenerator {
     new ConstantCallGen(DOUBLE_TYPE_INFO, Math.PI.toString))
 
   addSqlFunction(
+    E,
+    Seq(),
+    new ConstantCallGen(DOUBLE_TYPE_INFO, Math.E.toString))
+
+  addSqlFunction(
     RAND,
     Seq(),
     new RandCallGen(isRandInteger = false, hasSeed = false))
@@ -396,10 +402,17 @@ object FunctionGenerator {
     Seq(INT_TYPE_INFO, INT_TYPE_INFO),
     new RandCallGen(isRandInteger = true, hasSeed = true))
 
-  addSqlFunction(
-    E,
-    Seq(),
-    new ConstantCallGen(DOUBLE_TYPE_INFO, Math.E.toString))
+  addSqlFunctionMethod(
+    ScalarSqlFunctions.LOG,
+    Seq(DOUBLE_TYPE_INFO),
+    DOUBLE_TYPE_INFO,
+    BuiltInMethods.LOG)
+
+  addSqlFunctionMethod(
+    ScalarSqlFunctions.LOG,
+    Seq(DOUBLE_TYPE_INFO, DOUBLE_TYPE_INFO),
+    DOUBLE_TYPE_INFO,
+    BuiltInMethods.LOG_WITH_BASE)
 
   // ----------------------------------------------------------------------------------------------
   // Temporal functions

http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
index d27efbe..4cac1f7 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
@@ -21,7 +21,7 @@ import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
 import org.apache.calcite.sql.`type`.{OperandTypes, ReturnTypes, SqlTypeFamily}
 
 /**
-  * All build-in scalar sql functions.
+  * All built-in scalar SQL functions.
   */
 object ScalarSqlFunctions {
 
@@ -48,4 +48,14 @@ object ScalarSqlFunctions {
     null,
     OperandTypes.ONE_OR_MORE,
     SqlFunctionCategory.STRING)
+
+  val LOG = new SqlFunction(
+    "LOG",
+    SqlKind.OTHER_FUNCTION,
+    ReturnTypes.DOUBLE_NULLABLE,
+    null,
+    OperandTypes.or(OperandTypes.NUMERIC,
+      OperandTypes.family(SqlTypeFamily.NUMERIC, SqlTypeFamily.NUMERIC)),
+    SqlFunctionCategory.NUMERIC)
+
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
index 865421f..36edba2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/functions/ScalarFunctions.scala
@@ -82,4 +82,31 @@ object ScalarFunctions {
     }
     sb.toString
   }
+
+  /**
+    * Returns the natural logarithm of "x".
+    */
+  def log(x: Double): Double = {
+    if (x <= 0.0) {
+      throw new IllegalArgumentException(s"x of 'log(x)' must be > 0, but x = $x")
+    }
+    else {
+      Math.log(x)
+    }
+  }
+
+  /**
+    * Returns the logarithm of "x" with base "base".
+    */
+  def log(base: Double, x: Double): Double = {
+    if (x <= 0.0) {
+      throw new IllegalArgumentException(s"x of 'log(base, x)' must be > 0, but x = $x")
+    }
+    if (base <= 1.0) {
+      throw new IllegalArgumentException(s"base of 'log(base, x)' must be > 1, but base
= $base")
+    }
+    else {
+      Math.log(x) / Math.log(base)
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
index 36fcab1..6858135 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
@@ -405,12 +405,13 @@ class BasicOperatorTable extends ReflectiveSqlOperatorTable {
     SqlStdOperatorTable.SIGN,
     SqlStdOperatorTable.ROUND,
     SqlStdOperatorTable.PI,
+    ScalarSqlFunctions.E,
     SqlStdOperatorTable.RAND,
     SqlStdOperatorTable.RAND_INTEGER,
-    ScalarSqlFunctions.E,
     ScalarSqlFunctions.CONCAT,
     ScalarSqlFunctions.CONCAT_WS,
     SqlStdOperatorTable.TIMESTAMP_ADD,
+    ScalarSqlFunctions.LOG,
 
     // EXTENSIONS
     SqlStdOperatorTable.TUMBLE,

http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
index d477026..6a21f21 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
@@ -1188,6 +1188,34 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
       math.E.toString)
   }
 
+  @Test
+  def testLog(): Unit = {
+    testSqlApi(
+      "LOG(f6)",
+      "1.5260563034950492"
+    )
+
+    testSqlApi(
+      "LOG(f6-f6 + 10, f6-f6+100)",
+      "2.0"
+    )
+
+    testSqlApi(
+      "LOG(f6+20)",
+      "3.202746442938317"
+    )
+
+    testSqlApi(
+      "LOG(10)",
+      "2.302585092994046"
+    )
+
+    testSqlApi(
+      "LOG(10, 100)",
+      "2.0"
+    )
+  }
+
   // ----------------------------------------------------------------------------------------------
   // Temporal functions
   // ----------------------------------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/flink/blob/61d87d7b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/validation/ScalarFunctionsValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/validation/ScalarFunctionsValidationTest.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/validation/ScalarFunctionsValidationTest.scala
index 97b03d2..581799b 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/validation/ScalarFunctionsValidationTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/validation/ScalarFunctionsValidationTest.scala
@@ -26,6 +26,28 @@ import org.junit.Test
 class ScalarFunctionsValidationTest extends ScalarTypesTestBase {
 
   // ----------------------------------------------------------------------------------------------
+  // Math functions
+  // ----------------------------------------------------------------------------------------------
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testInvalidLog1(): Unit = {
+    // invalid arithmetic argument
+    testSqlApi(
+      "LOG(1, 100)",
+      "FAIL"
+    )
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testInvalidLog2(): Unit ={
+    // invalid arithmetic argument
+    testSqlApi(
+      "LOG(-1)",
+      "FAIL"
+    )
+  }
+
+  // ----------------------------------------------------------------------------------------------
   // String functions
   // ----------------------------------------------------------------------------------------------
 


Mime
View raw message