You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2019/01/08 11:03:36 UTC
[GitHub] asfgit closed pull request #7328: [FLINK-11191][table] Check for
ambiguous columns in MATCH_RECOGNIZE
asfgit closed pull request #7328: [FLINK-11191][table] Check for ambiguous columns in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7328
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamMatchRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamMatchRule.scala
index bc0f56e38ba..046a49aefdf 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamMatchRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/DataStreamMatchRule.scala
@@ -18,12 +18,15 @@
package org.apache.flink.table.plan.rules.datastream
+import java.util.{List => JList}
+
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.convert.ConverterRule
-import org.apache.calcite.rex.{RexCall, RexNode}
+import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode}
import org.apache.calcite.sql.SqlAggFunction
-import org.apache.flink.table.api.TableException
+import org.apache.flink.table.api.{TableException, ValidationException}
import org.apache.flink.table.plan.logical.MatchRecognize
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.nodes.datastream.DataStreamMatch
@@ -33,6 +36,7 @@ import org.apache.flink.table.plan.util.RexDefaultVisitor
import org.apache.flink.table.util.MatchUtil
import scala.collection.JavaConverters._
+import scala.collection.mutable
class DataStreamMatchRule
extends ConverterRule(
@@ -46,6 +50,8 @@ class DataStreamMatchRule
validateAggregations(logicalMatch.getMeasures.values().asScala)
validateAggregations(logicalMatch.getPatternDefinitions.values().asScala)
+ // This check might be obsolete once CALCITE-2747 is resolved
+ validateAmbiguousColumns(logicalMatch)
true
}
@@ -91,7 +97,48 @@ class DataStreamMatchRule
expr.foreach(_.accept(validator))
}
- class AggregationsValidator extends RexDefaultVisitor[Object] {
+ private def validateAmbiguousColumns(logicalMatch: FlinkLogicalMatch): Unit = {
+ if (logicalMatch.isAllRows) {
+ throw new TableException("All rows per match mode is not supported yet.")
+ } else {
+ val refNameFinder = new RefNameFinder(logicalMatch.getInput.getRowType)
+ validateAmbiguousColumnsOnRowPerMatch(
+ logicalMatch.getPartitionKeys,
+ logicalMatch.getMeasures.keySet().asScala,
+ logicalMatch.getRowType,
+ refNameFinder)
+ }
+ }
+
+ private def validateAmbiguousColumnsOnRowPerMatch(
+ partitionKeys: JList[RexNode],
+ measuresNames: mutable.Set[String],
+ expectedSchema: RelDataType,
+ refNameFinder: RefNameFinder)
+ : Unit = {
+ val actualSize = partitionKeys.size() + measuresNames.size
+ val expectedSize = expectedSchema.getFieldCount
+ if (actualSize != expectedSize) {
+ //try to find ambiguous column
+
+ val ambiguousColumns = partitionKeys.asScala.map(_.accept(refNameFinder))
+ .filter(measuresNames.contains).mkString("{", ", ", "}")
+
+ throw new ValidationException(s"Columns ambiguously defined: $ambiguousColumns")
+ }
+ }
+
+ private class RefNameFinder(inputSchema: RelDataType) extends RexDefaultVisitor[String] {
+
+ override def visitInputRef(inputRef: RexInputRef): String = {
+ inputSchema.getFieldList.get(inputRef.getIndex).getName
+ }
+
+ override def visitNode(rexNode: RexNode): String =
+ throw new TableException(s"PARTITION BY clause accepts only input reference. Found $rexNode")
+ }
+
+ private class AggregationsValidator extends RexDefaultVisitor[Object] {
override def visitCall(call: RexCall): AnyRef = {
call.getOperator match {
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/match/MatchRecognizeValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/match/MatchRecognizeValidationTest.scala
index b77a60e1af5..4e6562da281 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/match/MatchRecognizeValidationTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/match/MatchRecognizeValidationTest.scala
@@ -174,6 +174,30 @@ class MatchRecognizeValidationTest extends TableTestBase {
streamUtils.tableEnv.sqlQuery(sqlQuery).toAppendStream[Row]
}
+ @Test
+ def testValidatingAmbiguousColumns(): Unit = {
+ thrown.expectMessage("Columns ambiguously defined: {symbol, price}")
+ thrown.expect(classOf[ValidationException])
+
+ val sqlQuery =
+ s"""
+ |SELECT *
+ |FROM Ticker
+ |MATCH_RECOGNIZE (
+ | PARTITION BY symbol, price
+ | ORDER BY proctime
+ | MEASURES
+ | A.symbol AS symbol,
+ | A.price AS price
+ | PATTERN (A)
+ | DEFINE
+ | A AS symbol = 'a'
+ |) AS T
+ |""".stripMargin
+
+ streamUtils.tableEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ }
+
// ***************************************************************************************
// * Those validations are temporary. We should remove those tests once we support those *
// * features. *
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services