[FLINK-22523][table-planner-blink] Window TVF should throw helpful exception when specifying offset parameter (#15803)

wuchong · wuchong · commit 1088b8726732 · 2021-05-11T10:20:27.000+08:00
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/SqlCumulateTableFunction.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/SqlCumulateTableFunction.java
@@ -57,6 +57,9 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail
             if (!checkIntervalOperands(callBinding, 2)) {
                 return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);
             }
+            if (callBinding.getOperandCount() == 5) {
+                return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);
+            }
             // check time attribute
             return throwExceptionOrReturnFalse(
                     checkTimeColumnDescriptorOperand(callBinding, 1), throwOnFailure);
@@ -66,7 +69,7 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail
         public String getAllowedSignatures(SqlOperator op, String opName) {
             return opName
                     + "(TABLE table_name, DESCRIPTOR(timecol), "
-                    + "datetime interval, datetime interval[, datetime interval])";
+                    + "datetime interval, datetime interval)";
         }
     }
 }
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/SqlHopTableFunction.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/SqlHopTableFunction.java
@@ -57,6 +57,9 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail
             if (!checkIntervalOperands(callBinding, 2)) {
                 return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);
             }
+            if (callBinding.getOperandCount() == 5) {
+                return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);
+            }
             // check time attribute
             return throwExceptionOrReturnFalse(
                     checkTimeColumnDescriptorOperand(callBinding, 1), throwOnFailure);
@@ -66,7 +69,7 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail
         public String getAllowedSignatures(SqlOperator op, String opName) {
             return opName
                     + "(TABLE table_name, DESCRIPTOR(timecol), "
-                    + "datetime interval, datetime interval[, datetime interval])";
+                    + "datetime interval, datetime interval)";
         }
     }
 }
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/SqlTumbleTableFunction.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/SqlTumbleTableFunction.java
@@ -55,16 +55,17 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail
             if (!checkIntervalOperands(callBinding, 2)) {
                 return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);
             }
+            if (callBinding.getOperandCount() == 4) {
+                return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);
+            }
             // check time attribute
             return throwExceptionOrReturnFalse(
                     checkTimeColumnDescriptorOperand(callBinding, 1), throwOnFailure);
         }
 
         @Override
         public String getAllowedSignatures(SqlOperator op, String opName) {
-            return opName
-                    + "(TABLE table_name, DESCRIPTOR(timecol), datetime interval"
-                    + "[, datetime interval])";
+            return opName + "(TABLE table_name, DESCRIPTOR(timecol), datetime interval)";
         }
     }
 }
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/WindowTableFunctionTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/WindowTableFunctionTest.scala
@@ -140,4 +140,59 @@ class WindowTableFunctionTest extends TableTestBase {
     util.verifyExplain(sql)
   }
 
+  @Test
+  def testInvalidTumbleParameters(): Unit = {
+    val sql =
+      """
+        |SELECT *
+        |FROM TABLE(TUMBLE(
+        |   TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE, INTERVAL '5' MINUTE))
+        |""".stripMargin
+
+    thrown.expectMessage("Supported form(s): " +
+      "TUMBLE(TABLE table_name, DESCRIPTOR(timecol), datetime interval)")
+    thrown.expect(classOf[ValidationException])
+    util.verifyExplain(sql)
+  }
+
+  @Test
+  def testInvalidHopParameters(): Unit = {
+    val sql =
+      """
+        |SELECT *
+        |FROM TABLE(
+        |  HOP(
+        |    TABLE MyTable,
+        |    DESCRIPTOR(rowtime),
+        |    INTERVAL '1' MINUTE,
+        |    INTERVAL '15' MINUTE,
+        |    INTERVAL '5' MINUTE))
+        |""".stripMargin
+
+    thrown.expectMessage("Supported form(s): " +
+      "HOP(TABLE table_name, DESCRIPTOR(timecol), datetime interval, datetime interval)")
+    thrown.expect(classOf[ValidationException])
+    util.verifyExplain(sql)
+  }
+
+  @Test
+  def testInvalidCumulateParameters(): Unit = {
+    val sql =
+      """
+        |SELECT *
+        |FROM TABLE(
+        |  CUMULATE(
+        |    TABLE MyTable,
+        |    DESCRIPTOR(rowtime),
+        |    INTERVAL '1' MINUTE,
+        |    INTERVAL '15' MINUTE,
+        |    INTERVAL '5' MINUTE))
+        |""".stripMargin
+
+    thrown.expectMessage("Supported form(s): " +
+      "CUMULATE(TABLE table_name, DESCRIPTOR(timecol), datetime interval, datetime interval)")
+    thrown.expect(classOf[ValidationException])
+    util.verifyExplain(sql)
+  }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -57,6 +57,9 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail`
`57`	`57`	`if (!checkIntervalOperands(callBinding, 2)) {`
`58`	`58`	`return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);`
`59`	`59`	`}`
	`60`	`+ if (callBinding.getOperandCount() == 5) {`
	`61`	`+ return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);`
	`62`	`+ }`
`60`	`63`	`// check time attribute`
`61`	`64`	`return throwExceptionOrReturnFalse(`
`62`	`65`	`checkTimeColumnDescriptorOperand(callBinding, 1), throwOnFailure);`
`@@ -66,7 +69,7 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail`
`66`	`69`	`public String getAllowedSignatures(SqlOperator op, String opName) {`
`67`	`70`	`return opName`
`68`	`71`	`+ "(TABLE table_name, DESCRIPTOR(timecol), "`
`69`		`- + "datetime interval, datetime interval[, datetime interval])";`
	`72`	`+ + "datetime interval, datetime interval)";`
`70`	`73`	`}`
`71`	`74`	`}`
`72`	`75`	`}`
Original file line number	Diff line number	Diff line change
`@@ -55,16 +55,17 @@ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFail`
`55`	`55`	`if (!checkIntervalOperands(callBinding, 2)) {`
`56`	`56`	`return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);`
`57`	`57`	`}`
	`58`	`+ if (callBinding.getOperandCount() == 4) {`
	`59`	`+ return throwValidationSignatureErrorOrReturnFalse(callBinding, throwOnFailure);`
	`60`	`+ }`
`58`	`61`	`// check time attribute`
`59`	`62`	`return throwExceptionOrReturnFalse(`
`60`	`63`	`checkTimeColumnDescriptorOperand(callBinding, 1), throwOnFailure);`
`61`	`64`	`}`
`62`	`65`
`63`	`66`	`@Override`
`64`	`67`	`public String getAllowedSignatures(SqlOperator op, String opName) {`
`65`		`- return opName`
`66`		`- + "(TABLE table_name, DESCRIPTOR(timecol), datetime interval"`
`67`		`- + "[, datetime interval])";`
	`68`	`+ return opName + "(TABLE table_name, DESCRIPTOR(timecol), datetime interval)";`
`68`	`69`	`}`
`69`	`70`	`}`
`70`	`71`	`}`