Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3166#discussion_r102507787
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/TableSourceTest.scala
---
@@ -126,21 +156,49 @@ class TableSourceTest extends TableTestBase {
@Test
def testStreamProjectableSourceScanNoIdentityCalc(): Unit = {
- val (csvTable, tableName) = tableSource
+ val (tableSource, tableName) = csvTable
val util = streamTestUtil()
val tEnv = util.tEnv
- tEnv.registerTableSource(tableName, csvTable)
+ tEnv.registerTableSource(tableName, tableSource)
val result = tEnv
.scan(tableName)
.select('id, 'score, 'first)
- val expected = sourceStreamTableNode(tableName, noCalcFields)
+ val expected = projectableSourceStreamTableNode(tableName,
noCalcFields)
util.verifyTable(result, expected)
}
@Test
+ def testStreamFilterableSourceScanPlanTableApi(): Unit = {
+ val (tableSource, tableName) = filterableTableSource
+ val util = streamTestUtil()
+ val tEnv = util.tEnv
+
+ tEnv.registerTableSource(tableName, tableSource)
+
+ val result = tEnv
+ .scan(tableName)
+ .select('price, 'id, 'amount)
+ .where("amount > 2 && price * 2 < 32")
--- End diff --
An example for an unsupported predicate would be
`'id.cast(BasicTypeInfo.STRING_TYPE_INFO) === "abc"`. This throws and exception
when translating it to an `Expression`.
As said before, unsupported expressions should be gracefully handled by not
failing but instead by not offering this `RexNode` to the
`FilterableTableSource` and evaluating it in the `DataSetCalc`.
I would suggest to use `CAST` as an example to implement the graceful
handling and adding support for it once the failure-free translation works.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
