[FLINK-8868] [table] Support Table Function as Table Source for Stream Sql #6574

Xpray · 2018-08-17T04:51:03Z

What is the purpose of the change

Support Table Function as Table source for Stream Sql

TableFunction might produce infinite records, hence the support for batch sql should be discussed.

Brief change log

Add new DataStreamTableFunctionScan

Verifying this change

This change added tests and can be verified as follows:
new test cases in:
org.apache.flink.table.runtime.stream.sql.SqlITCase

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? no

alpinegizmo · 2018-08-17T06:32:33Z

FLINK-8688 is "Enable distinct aggregation for data stream on Table/SQL API", which doesn't seem related to this PR.

Xpray · 2018-08-17T07:16:06Z

It should be FLINK-8868, I'll fix this.

pnowojski

Thanks for the contribution! I left couple of comments in the code.

pnowojski · 2018-08-20T12:43:49Z

...braries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/SqlITCase.scala

+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env)
+    StreamITCase.clear
+    tEnv.registerFunction("udtf", new TableFunc2WithBase)


Do we need this TableFunc2WithBase here? Couldn't we use TableFunc2? What does TableFunc2WithBase give us?

pnowojski · 2018-08-20T12:51:44Z

...braries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/SqlITCase.scala

@@ -897,6 +897,45 @@ class SqlITCase extends StreamingWithStateTestBase {

    assertEquals(List(expected.toString()), StreamITCase.testResults.sorted)
  }
+
+  @Test
+  def tableFunctionAsSource(): Unit = {


I think this file is already too big and at least this new tests should be put into something like TableFunctionITCase.

pnowojski · 2018-08-20T13:11:04Z

...ain/scala/org/apache/flink/table/plan/rules/datastream/DataStreamTableFunctionScanRule.scala

+    val scan: FlinkLogicalTableFunctionScan = rel.asInstanceOf[FlinkLogicalTableFunctionScan]
+    val traitSet = rel.getTraitSet.replace(FlinkConventions.DATASTREAM)
+
+    new DataStreamTableFunctionScan(


Should this conversion happen always? How does it play along with for example LogicalCorrelate nodes, DataStreamCorrelateRule and org.apache.flink.table.api.stream.sql.CorrelateTest?

Does it work because if vulcano planer picks this rule in case of LogicalCorrelate it is later unable to convert LogicalCorrelate to DataStreamCorrelate and thus retracts from DataStreamTableFunctionScanRule?

There might be two cases that TableFunction acts like a Source, one is 'Select * from Lateral Table(udtf())', the other is that the left table do not correlate with the right table so there's no LogicalCorrelate node.

pnowojski · 2018-08-20T13:13:50Z

...braries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/SqlITCase.scala

@@ -897,6 +897,45 @@ class SqlITCase extends StreamingWithStateTestBase {

    assertEquals(List(expected.toString()), StreamITCase.testResults.sorted)
  }
+
+  @Test


Please add tests for table API as well. I would expect there to have this error:

org.apache.flink.table.api.ValidationException: Cannot translate a query with an unbounded table function call. at org.apache.flink.table.api.Table.getRelNode(table.scala:94) at org.apache.flink.table.utils.StreamTableTestUtil.printTable(TableTestBase.scala:321)

I add a test case for TableAPI and a ValidationException with "TableFunction can only be used in join and leftOuterJoin" will be thrown.

Sorry, I meant the other way around. We should try to fix this for table api. By saying:

I would expect there to have this error:

I didn't mean that "I would like to have test asserting this validation exception", but "I think you missed testing this feature on table API and it probably will fail there with validation exception"

@pnowojski , I think it's better to support SQL only for this time. TableAPI needs more effort.

What's the problem with Table API here? I had a suspicion that this ValidationException:

def getRelNode: RelNode = if (containsUnboundedUDTFCall(logicalPlan)) { throw new ValidationException("Cannot translate a query with an unbounded table function call.") } else { logicalPlan.toRelNode(relBuilder) }

is being thrown mostly as a precaution, since previously there was no execution code to support it. Now (with this PR) that will not be the case anymore. What would happen if we simply removed it?

pnowojski · 2018-08-20T13:38:39Z

.../flink-table/src/main/scala/org/apache/flink/table/codegen/SourceFunctionCodeGenerator.scala

+         |
+         |  }
+         |
+         |  class ${collectorName} implements ${classOf[Collector[_]].getCanonicalName} {


Isn't this duplicated with org.apache.flink.table.plan.nodes.CommonCorrelate#generateCollector?

pnowojski · 2018-08-20T13:39:44Z

.../flink-table/src/main/scala/org/apache/flink/table/codegen/SourceFunctionCodeGenerator.scala

+
+    val functionCode =
+      s"""
+         |public class $funcName extends ${classOf[RichSourceFunction[_]].getCanonicalName} {


And doesn't this share a lot of code with org.apache.flink.table.plan.nodes.CommonCorrelate#generateFunction?

I'll do some refactor here.

pnowojski · 2018-08-20T13:56:06Z

.../flink-table/src/main/scala/org/apache/flink/table/codegen/SourceFunctionCodeGenerator.scala

+  outputType: TypeInformation[T])
+  extends CodeGenerator(config, false, new RowTypeInfo(), None, None) {
+
+  def generateSourceFunction(


Sorry for asking maybe a stupid question. Why do we need to generate so much code? Should most of the code here (all even all of it) be in standard scala/java classes and the only thing that should be generate is a RexCall as as an implementation of some interface, that should be used by some concrete java/scala class?

In other words, generating implementation of RichSourceFunction seams like a bit an overkill.

I think the UDTF might produces infinite records and can do some initialization, which makes it look like a real source.

pnowojski · 2018-08-21T12:06:15Z

@twalthr could you take a look at this one? Especially at the code generation parts since I have little to no experience in this regard.

…m Sql

Xpray force-pushed the FLINK-8688 branch from 0f7ad61 to d3aca91 Compare August 17, 2018 04:55

Xpray changed the title ~~[FLINK-8688] [table] Support Table Function as Table for Stream Sql~~ [FLINK-8688] [table] Support Table Function as Table Source for Stream Sql Aug 17, 2018

Xpray force-pushed the FLINK-8688 branch 2 times, most recently from 998834d to feaf163 Compare August 17, 2018 06:13

Xpray changed the title ~~[FLINK-8688] [table] Support Table Function as Table Source for Stream Sql~~ [FLINK-8868] [table] Support Table Function as Table Source for Stream Sql Aug 17, 2018

Xpray force-pushed the FLINK-8688 branch 2 times, most recently from e6b2fcf to 94f7da2 Compare August 17, 2018 14:50

pnowojski requested changes Aug 20, 2018

View reviewed changes

[FLINK-8868] [table] Support Table Function as Table Source for Strea…

3e3de47

…m Sql

Xpray force-pushed the FLINK-8688 branch from 94f7da2 to 3e3de47 Compare August 21, 2018 13:39

KurtYoung closed this Jan 30, 2019

rmetzger added the component=API/TableSQL label Mar 18, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-8868] [table] Support Table Function as Table Source for Stream Sql #6574

[FLINK-8868] [table] Support Table Function as Table Source for Stream Sql #6574

Xpray commented Aug 17, 2018 •

edited

Loading

alpinegizmo commented Aug 17, 2018

Xpray commented Aug 17, 2018

pnowojski left a comment

pnowojski Aug 20, 2018

pnowojski Aug 20, 2018

pnowojski Aug 20, 2018

Xpray Aug 20, 2018 •

edited

Loading

pnowojski Aug 20, 2018

Xpray Aug 21, 2018

pnowojski Aug 22, 2018

Xpray Aug 22, 2018

pnowojski Aug 23, 2018

pnowojski Aug 20, 2018

pnowojski Aug 20, 2018

Xpray Aug 20, 2018

pnowojski Aug 20, 2018

Xpray Aug 20, 2018

pnowojski commented Aug 21, 2018

[FLINK-8868] [table] Support Table Function as Table Source for Stream Sql #6574

[FLINK-8868] [table] Support Table Function as Table Source for Stream Sql #6574

Conversation

Xpray commented Aug 17, 2018 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

alpinegizmo commented Aug 17, 2018

Xpray commented Aug 17, 2018

pnowojski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Xpray Aug 20, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski commented Aug 21, 2018

Xpray commented Aug 17, 2018 •

edited

Loading

Xpray Aug 20, 2018 •

edited

Loading