[FLINK-6893] [table] Add BIN function support

This closes apache#4128.
G356 · Jan 10, 2018 · 2914e59 · 2914e59
1 parent d42759d
commit 2914e59
Show file tree

Hide file tree

Showing 7 changed files with 63 additions and 4 deletions.
diff --git a/docs/dev/table/sql.md b/docs/dev/table/sql.md
@@ -1746,6 +1746,16 @@ CONCAT_WS(separator, string1, string2,...)
         <p>Returns the string that results from concatenating the arguments using a separator. The separator is added between the strings to be concatenated. Returns NULL If the separator is NULL. CONCAT_WS() does not skip empty strings. However, it does skip any NULL argument. E.g. <code>CONCAT_WS("~", "AA", "BB", "", "CC")</code> returns <code>AA~BB~~CC</code></p>
       </td>
     </tr>
+    <tr>
+      <td>
+{% highlight text %}
+BIN(numeric)
+      {% endhighlight %}
+      </td>
+      <td>
+        <p>Returns a string representation of the binary value of Numeric, Returns NULL if Numeric is NULL.</p>
+      </td>
+    </tr>
 
   </tbody>
 </table>

diff --git a/...ries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala b/...ries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/BuiltInMethods.scala
@@ -96,4 +96,5 @@ object BuiltInMethods {
     Types.lookupMethod(
       classOf[ScalarFunctions], "concat_ws", classOf[String], classOf[Array[String]])
 
+  val BIN = Types.lookupMethod(classOf[JLong], "toBinaryString", classOf[Long])
 }
diff --git a/...s/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala b/...s/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala
@@ -413,6 +413,11 @@ object FunctionGenerator {
     DOUBLE_TYPE_INFO,
     BuiltInMethods.LOG_WITH_BASE)
 
+  addSqlFunction(
+    ScalarSqlFunctions.E,
+    Seq(),
+    new ConstantCallGen(DOUBLE_TYPE_INFO, Math.E.toString))
+
   // ----------------------------------------------------------------------------------------------
   // Temporal functions
   // ----------------------------------------------------------------------------------------------
@@ -502,6 +507,12 @@ object FunctionGenerator {
     new DateFormatCallGen
   )
 
+  addSqlFunctionMethod(
+    ScalarSqlFunctions.BIN,
+    Seq(LONG_TYPE_INFO),
+    STRING_TYPE_INFO,
+    BuiltInMethods.BIN)
+
   // ----------------------------------------------------------------------------------------------
   // Cryptographic Hash functions
   // ----------------------------------------------------------------------------------------------

diff --git a/.../flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala b/.../flink-table/src/main/scala/org/apache/flink/table/functions/sql/ScalarSqlFunctions.scala
@@ -15,6 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.flink.table.functions.sql
 
 import org.apache.calcite.sql.{SqlFunction, SqlFunctionCategory, SqlKind}
@@ -33,6 +34,14 @@ object ScalarSqlFunctions {
     OperandTypes.NILADIC,
     SqlFunctionCategory.NUMERIC)
 
+  val BIN = new SqlFunction(
+    "BIN",
+    SqlKind.OTHER_FUNCTION,
+    ReturnTypes.explicit(SqlTypeName.VARCHAR),
+    null,
+    OperandTypes.family(SqlTypeFamily.INTEGER),
+    SqlFunctionCategory.NUMERIC)
+
   val CONCAT = new SqlFunction(
     "CONCAT",
     SqlKind.OTHER_FUNCTION,

diff --git a/...ibraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala b/...ibraries/flink-table/src/main/scala/org/apache/flink/table/validate/FunctionCatalog.scala
@@ -416,6 +416,7 @@ class BasicOperatorTable extends ReflectiveSqlOperatorTable {
     SqlStdOperatorTable.RAND_INTEGER,
     ScalarSqlFunctions.CONCAT,
     ScalarSqlFunctions.CONCAT_WS,
+    ScalarSqlFunctions.BIN,
     SqlStdOperatorTable.TIMESTAMP_ADD,
     ScalarSqlFunctions.LOG,
     ScalarSqlFunctions.MD5,

diff --git a/...s/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala b/...s/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
@@ -352,6 +352,18 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
       "Flink~~~~xx")
   }
 
+  @Test
+  def testBin(): Unit = {
+    testSqlApi("BIN(f2)", "101010")
+    testSqlApi("BIN(f3)", "101011")
+    testSqlApi("BIN(f4)", "101100")
+    testSqlApi("BIN(f7)", "11")
+    testSqlApi("BIN(12)", "1100")
+    testSqlApi("BIN(10)", "1010")
+    testSqlApi("BIN(0)", "0")
+    testSqlApi("BIN(f32)","1111111111111111111111111111111111111111111111111111111111111111")
+  }
+
   // ----------------------------------------------------------------------------------------------
   // Math functions
   // ----------------------------------------------------------------------------------------------

diff --git a/...t/scala/org/apache/flink/table/expressions/validation/ScalarFunctionsValidationTest.scala b/...t/scala/org/apache/flink/table/expressions/validation/ScalarFunctionsValidationTest.scala
@@ -39,7 +39,7 @@ class ScalarFunctionsValidationTest extends ScalarTypesTestBase {
   }
 
   @Test(expected = classOf[IllegalArgumentException])
-  def testInvalidLog2(): Unit ={
+  def testInvalidLog2(): Unit = {
     // invalid arithmetic argument
     testSqlApi(
       "LOG(-1)",
@@ -68,17 +68,17 @@ class ScalarFunctionsValidationTest extends ScalarTypesTestBase {
   // ----------------------------------------------------------------------------------------------
 
   @Test(expected = classOf[SqlParserException])
-  def testTimestampAddWithWrongTimestampInterval(): Unit ={
+  def testTimestampAddWithWrongTimestampInterval(): Unit = {
     testSqlApi("TIMESTAMPADD(XXX, 1, timestamp '2016-02-24'))", "2016-06-16")
   }
 
   @Test(expected = classOf[SqlParserException])
-  def testTimestampAddWithWrongTimestampFormat(): Unit ={
+  def testTimestampAddWithWrongTimestampFormat(): Unit = {
     testSqlApi("TIMESTAMPADD(YEAR, 1, timestamp '2016-02-24'))", "2016-06-16")
   }
 
   @Test(expected = classOf[ValidationException])
-  def testTimestampAddWithWrongQuantity(): Unit ={
+  def testTimestampAddWithWrongQuantity(): Unit = {
     testSqlApi("TIMESTAMPADD(YEAR, 1.0, timestamp '2016-02-24 12:42:25')", "2016-06-16")
   }
 
@@ -112,4 +112,19 @@ class ScalarFunctionsValidationTest extends ScalarTypesTestBase {
       "true"
     )
   }
+
+  @Test(expected = classOf[ValidationException])
+  def testInvalidBin1(): Unit = {
+    testSqlApi("BIN(f12)", "101010") // float type
+  }
+
+  @Test(expected = classOf[ValidationException])
+  def testInvalidBin2(): Unit = {
+    testSqlApi("BIN(f15)", "101010") // BigDecimal type
+  }
+
+  @Test(expected = classOf[ValidationException])
+  def testInvalidBin3(): Unit = {
+    testSqlApi("BIN(f16)", "101010") // Date type
+  }
 }