diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
index a1228fc534071..b303da49fb219 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
@@ -617,6 +617,17 @@ public SqlMonotonicity getMonotonicity(SqlOperatorBinding call) {
 		),
 		SqlFunctionCategory.STRING);
 
+	public static final SqlFunction SUBSTR = new SqlFunction(
+			"SUBSTR",
+			SqlKind.OTHER_FUNCTION,
+			ARG0_VARCHAR_FORCE_NULLABLE,
+			null,
+			OperandTypes.or(
+					OperandTypes.family(SqlTypeFamily.CHARACTER, SqlTypeFamily.INTEGER),
+					OperandTypes.family(SqlTypeFamily.CHARACTER, SqlTypeFamily.INTEGER, SqlTypeFamily.INTEGER)
+			),
+			SqlFunctionCategory.STRING);
+
 	public static final SqlFunction LEFT = new SqlFunction(
 		"LEFT",
 		SqlKind.OTHER_FUNCTION,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala
index 0e7c5d1716ff7..7f40f73ab59f0 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala
@@ -63,7 +63,7 @@ object StringCallGen {
       case NOT_LIKE =>
         generateNot(ctx, new LikeCallGen().generate(ctx, operands, new BooleanType()))
 
-      case SUBSTRING => generateSubString(ctx, operands)
+      case SUBSTR | SUBSTRING => generateSubString(ctx, operands)
 
       case LEFT => generateLeft(ctx, operands.head, operands(1))
 
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala
index 921b2ace33731..7e094a4be92d8 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala
@@ -938,7 +938,7 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
 
   @Test
   def testSubString(): Unit = {
-    Array("substring").foreach {
+    Array("substring", "substr").foreach {
       substr =>
         testSqlApi(s"$substr(f0, 2, 3)", "his")
         testSqlApi(s"$substr(f0, 2, 100)", "his is a test String.")
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/SqlExpressionTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/SqlExpressionTest.scala
index 42de09d39c3ea..53e5f7c604ec1 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/SqlExpressionTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/SqlExpressionTest.scala
@@ -158,6 +158,10 @@ class SqlExpressionTest extends ExpressionTestBase {
     testSqlApi("SUBSTRING('hello world', 2)", "ello world")
     testSqlApi("SUBSTRING('hello world', 2, 3)", "ell")
     testSqlApi("SUBSTRING('hello world', 2, 300)", "ello world")
+    testSqlApi("SUBSTR('hello world', 2, 3)", "ell")
+    testSqlApi("SUBSTR('hello world', 2)", "ello world")
+    testSqlApi("SUBSTR('hello world', 2, 300)", "ello world")
+    testSqlApi("SUBSTR('hello world', 0, 3)", "hel")
     testSqlApi("INITCAP('hello world')", "Hello World")
     testSqlApi("REGEXP_REPLACE('foobar', 'oo|ar', '')", "fb")
     testSqlApi("REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 2)", "bar")