[FLINK-19976] FLIP-136: Improve interoperability between DataStream and Table API

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.14.0
• Component/s: Table SQL / API
• Assignee: Unassigned
• Reporter: Timo Walther
• Created: 04/Nov/20 15:07
• Updated: 07/Sep/21 07:31
• Resolved: 07/Sep/21 07:31

Description

The Table API has received many new features over the last year. It supports a new type system (FLIP-37), connectors support changelogs (FLIP-95), we have well defined internal data structures (FLIP-95), support for result retrieval in an interactive fashion (FLIP-84), and soon new TableDescriptors (FLIP-129).

However, the interfaces from and to DataStream API have not been touched during the introduction of these new features and are kind of outdated. The interfaces lack important functionality that is available in Table API but not exposed to DataStream API users. DataStream API is still our most important API which is why a good interoperability is crucial.

This FLIP is a mixture of different topics that improve the interoperability between DataStream and Table API in terms of:

• DataStream ↔ Table conversion
• translation of type systems TypeInformation ↔ DataType
• schema definition (incl. rowtime, watermarks, primary key)
• changelog handling
• row handling in DataStream API

See:
https://cwiki.apache.org/confluence/pages/resumedraft.action?draftId=158872537&draftShareId=69731f11-580f-4bfe-847f-e5eb3e585f1f&

Issue Links

• contains: FLINK-8545 Implement upsert stream table source
• mentioned in: Page Loading...

Sub-Tasks

• Support all conversion classes in Row.equals/hashCode: Closed
• Improve dealing with Row in DataStream API: Closed
• Add a converter from TypeInformation to DataType: Closed
• Allow SQL expressions in Table API: Closed
• Replace TableSchema with dedicated CatalogSchema: Closed
• Create a utility to create DataStream API's DataType and Schema: Closed
• Add new StreamTableEnvironment.fromDataStream: Closed
• Add new StreamTableEnvironment.toDataStream: Closed
• Allow casting between row and structured type: Closed
• Add StreamTableEnvironment.to/fromChangelogStream: Closed
• Make structured type's fields more lenient during casting: Closed
• Add Scala implicit conversions for new API methods: Closed
• Propagate Boundedness of SourceFunctionProvider to Transformation: Closed
• Disable AutoWatermarkInterval for bounded legacy sources: Closed
• Disable AllVerticesInSameSlotSharingGroupByDefault in DataStream batch mode: Closed
• Disable key sorting for Transformations created within table planner: Closed
• Support object reuse disabled in OperatorChain: Closed
• Expose a consistent GlobalDataExchangeMode: Closed
• Simplify BlinkExecutorFactory stack: Closed
• Use a layered configuration in Executor and Planner: Closed
• Use pipeline name consistently across DataStream API and Table API: Closed
• Use consistent managed memory weights for StreamNode: Closed
• Support DataStream batch mode in StreamTableEnvironment: Closed
• Add StreamStatementSet.attachAsDataStream: Closed
• Test StreamTableEnvironment batch mode in Python: Closed
• Deprecate toAppendStream and toRetractStream: Closed
• Type mismatch when declaring SOURCE_WATERMARK on TIMESTAMP_LTZ column: Closed
• Add documentation how to interact with DataStream API: Closed
• Align FLIP-136 (Improve interoperability between DataStream and Table API) in PyFlink Table API: Resolved
• Add documentation for batch mode in StreamTableEnvironment: Closed