[FLINK-18840] Add StreamStatementSet.attachAsDataStream

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.14.0
• Component/s: Table SQL / API
• Labels: pull-request-available usability
• Assignee: Timo Walther
• Reporter: Timo Walther
• Created: 06/Aug/20 12:54
• Updated: 16/Aug/21 06:59
• Resolved: 16/Aug/21 06:59

Description

StatementSet solves use cases for pure SQL & Table API pipelines. However, currently there is no way of creating StatementSet for a DataStream API job.

We propose the following API:

StreamTableEnvironment.createStatementSet(): StreamStatementSet // return a stream-specific set

StreamStatementSet extends StatementSet {
  /**
   * Attaches the optimized statement set to the DataStream pipeline.
   */
  attachAsDataStream(): Unit
}

An example could look like:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

tEnv
  .createStatementSet()
  .addInsert(tEnv.from("FromTable"))
  .attachAsDataStream();

tEnv.from("OtherTable").toDataStream(...) // continue with further operations

Issue Links

• links to: GitHub Pull Request #16816