[FLINK-16987] FLIP-95: Add new table source and sink interfaces

• Type: New Feature
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Component/s: Table SQL / API, Table SQL / Planner
• Labels: auto-deprioritized-major pull-request-available
• Assignee: Unassigned
• Reporter: Timo Walther
• Created: 06/Apr/20 07:08
• Updated: 29/Sep/23 22:35

Description

Proper support for handling changelogs, more efficient processing of data through the new Blink planner, and unified interfaces that are DataStream API agnostic make it necessary to rework the table source and sink interfaces.

The goals of this FLIP are:

• Simplify the current interface architecture:
  • Merge upsert, retract, and append sinks.
  • Unify batch and streaming sources.
  • Unify batch and streaming sinks.

• Allow sources to produce a changelog:
  • UpsertTableSources have been requested a lot by users. Now is the time to open the internal planner capabilities via the new interfaces.
  • According to FLIP-105, we would like to support changelogs for processing formats such as Debezium.

• Don't rely on DataStream API for source and sinks:
  • According to FLIP-32, the Table API and SQL should be independent of the DataStream API which is why the `table-common` module has no dependencies on `flink-streaming-java`.
  • Source and sink implementations should only depend on the `table-common` module after FLIP-27.
  • Until FLIP-27 is ready, we still put most of the interfaces in `table-common` and strictly separate interfaces that communicate with a planner and actual runtime reader/writers.

• Implement efficient sources and sinks without planner dependencies:
  • Make Blink's internal data structures available to connectors.
  • Introduce stable interfaces for data structures that can be marked as `@PublicEvolving`.

Issue Links

• mentioned in: Page Loading...

Sub-Tasks

• Add new data structure interfaces in table-common: Closed
• Add core table source/sink interfaces: Closed
• Add ability interfaces for table source/sink: Closed
• Add new factory interfaces and utilities: Closed
• Refactor planner and connectors to use new data structures: Closed
• Support ScanTableSource in planner: Closed
• Support LookupTableSource in planner: Closed
• Support DynamicTableSink in planner: Closed
• Fix shortcomings in new data structures: Closed
• Add a changeflag to Row type: Closed
• Data structure should cover all conversions declared in logical types: Closed
• Use new data structure converters when legacy types are not present: Closed
• Type information in sources should cover all data structures: Closed
• Refactor retraction rules to support inferring ChangelogMode: Closed
• Refactor BaseRow to use RowKind instead of byte header: Closed
• Support SupportsProjectionPushDown in planner: Closed
• Support SupportsFilterPushDown in planner: Closed
• Support SupportsLimitPushDown in planner: Closed
• Support SupportsPartitionPushDown in planner: Closed
• Support SupportsOverwrite in planner: Closed
• Support SupportsPartitioning in planner: Closed
• Support SupportsComputedColumnPushDown in planner: Closed
• Support SupportsWatermarkPushDown in planner: Closed
• SinkFormat#createSinkFormat should use DynamicTableSink.Context as the first parameter: Closed
• Improve new connector options exception in old planner: Closed
• Add createTypeInformation to DynamicTableSink#Context: Closed
• Remove deprecated RowData.get and similar methods: Closed
• Improve interface of ScanFormatFactory and SinkFormatFactory: Closed
• Add a table source example: Closed
• Add documentation for how to develop a new table source/sink: Closed
• Update the Row.toString method: Closed
• Support to integrate FLIP-27 source interface with Table API: Closed
• Support to bridge Transformation (DataStream) with FLIP-95 interface?: Closed
• Support the SupportsProjectionPushDown interface for LookupTableSource: Closed
• Support the SupportsFilterPushDown interface for LookupTableSource: Open
• Deprecate old source and sink interfaces: Closed
• Support nested push down for SupportsProjectionPushDown interface in planner: Closed
• Support the filter push down for the Jdbc connector: Closed
• Support the limit push down for the Jdbc connector: Closed
• Support the limit push down for the hbase connector: Open
• Support the nested projection push down for the hbase connector: Open
• Refactor and merge SupportsComputedColumnPushDown and SupportsWatermarkPushDown interfaces: Closed
• Support watermark push down with WatermarkStrategy: Closed
• Supports nested projection pushdown for filesystem connector of columnar format: Open
• Support Watermark push down for kafka connector: Closed
• Remove deprecated CatalogTable.getProperties: Closed
• predicates can not be pushed into TableScan when there is a WatermarkAssigner between Filter and Scan: Closed
• Add a SupportsSourceWatermark ability interface: Closed
• Allow prefix syntax for ConfigOption.mapType: Closed
• Consider ConfigOption fallback keys in FactoryUtil: Closed
• Validate partition columns for ResolvedCatalogTable: Closed
• Non-equality predicates on partition columns lead to incorrect plans: Closed
• Push down partitions before filters: Closed
• Clarify semantics of filter, projection, partition, and metadata pushdown: Open
• Add producedDataType to SupportsProjectionPushDown.applyProjection: Closed
• DynamicTableFactory.Context.getCatalogTable().getPartitionKeys() should return indexes: Open
• Metadata keys should not conflict with physical columns: Closed