[FLINK-15912][table] Add Context to TableSourceFactory and TableSinkFactory #11047

JingsongLi · 2020-02-10T08:01:19Z

What is the purpose of the change

Discussion in: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Improve-TableFactory-td36647.html

Vote in: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/VOTE-Improve-TableFactory-to-add-Context-td37211.html

Motivation:
Now the main needs and problems are:

Connector can't get TableConfig[1], and some behaviors really need to be
controlled by the user's table configuration. In the era of catalog, we
can't put these config in connector properties, which is too inconvenient.
A context class also allows for future modifications without touching the TableFactory interface again.

Brief change log

Interface modification for table factories
Apply new interface in blink planner.

Verifying this change

SinkTest.testTableSourceSinkFactory

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

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

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2020-02-10T08:03:11Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 86c4939 (Mon Feb 10 08:03:11 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-02-10T08:24:06Z

CI report:

86c4939 Travis: FAILURE Azure: FAILURE
18beb9a Travis: FAILURE Azure: SUCCESS
a9125de Travis: FAILURE Azure: FAILURE
52bc7dc Travis: FAILURE Azure: FAILURE
a307994 Travis: FAILURE Azure: FAILURE
4749ac0 Travis: FAILURE Azure: FAILURE
8c7c673 Travis: FAILURE Azure: PENDING
b1f4018 Travis: FAILURE Azure: SUCCESS
3428261 Travis: SUCCESS Azure: SUCCESS
2db8738 Travis: FAILURE Azure: FAILURE
daf7fda Travis: FAILURE
eba92d7 Travis: SUCCESS Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

wuchong · 2020-02-11T05:34:23Z

...able/flink-table-common/src/main/java/org/apache/flink/table/factories/TableSinkFactory.java

+		CatalogTable getTable();
+
+		/**
+		 * @return readable config of this table environment.


Add a description that the configuration gives the factory instance the ability to access TableConfig#getConfiguration() which holds the current TableEnvironment session configurations.

wuchong · 2020-02-11T05:35:14Z

...able/flink-table-common/src/main/java/org/apache/flink/table/factories/TableSinkFactory.java

+	 * @param context context of this table sink.
+	 * @return the configured table sink.
+	 */
+	default TableSink<T> createTableSink(Context context) {


Should we deprecate the other createTableSink and createTableSource interfaces?

I don't know... we can deprecate createTableSink(ObjectPath, CatalogTable). CC: @twalthr

wuchong · 2020-02-11T05:38:05Z

...le/flink-table-api-java/src/main/java/org/apache/flink/table/factories/TableFactoryUtil.java

 		return Optional.empty();
 	}

+	/**


We should make sure TableSinkFactory#createTableSink(context) is the only entry to be invoked by the planner. However, the above createTableSinkForCatalogTable method still calling the old createTableSink method in old planner.

I'll push another commit to Support create table source/sink by context in legacy planner

wuchong · 2020-02-11T05:51:04Z

...le-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SinkTest.scala

+  def testTableSourceSinkFactory(): Unit = {
+    val factory = new TestContextTableFactory
+    util.tableEnv.getConfig.getConfiguration.setBoolean(factory.needContain, true)
+    util.tableEnv.registerCatalog("cat", new GenericInMemoryCatalog("default") {


This only guarantees one of the paths, i.e. via TableFactoryUtil#createTableSinkForCatalogTable, which is only used by Hive. However, the another path is not covered, i.e. via CatalogSourceTable#findAndCreateTableSource which is used by most users.

Maybe you can upgrade the existing TestCollectionTableFactory to use the new interface and expose a session configuration, e.g. collection.is-bounded, the flag will be accessed via Context#getConfiguration and pass to the created CollectionTableSource.

Yes, you are right, but I want to modify this step by step.
The previous method createTableSink(ObjectPath tablePath, CatalogTable table) is only work in createTableSinkForCatalogTable too.
after these commits looks good to you, I will create following commit to fix this.
What do you think?

But without a test to cover the new interface, we have no idea whether it works as expected.

What do you mean? Is this not a test cover?

As I mentioned above, this test doesn't cover another path, i.e. CatalogSourceTable#findAndCreateTableSource.

I'll modify these things by multi commits.

JingsongLi · 2020-02-11T09:46:38Z

We can wait this one: #11055 , to be more cleaner.

wuchong

Thanks @JingsongLi . It looks good in general. I left some minor comments.

...link-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableFactoryTest.java

...able/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogTableImpl.java

...able/flink-table-common/src/main/java/org/apache/flink/table/factories/TableSinkFactory.java

wuchong · 2020-02-17T06:20:18Z

...api-java-bridge/src/main/java/org/apache/flink/table/factories/StreamTableSourceFactory.java

+	 */
+	@Override
+	default TableSource<T> createTableSource(Context context) {
+		return createStreamTableSource(context);


We should validate the return value of createStreamTableSource is not null. If is null, we should throw an exception to indicate users to implement createTableSource(context), because createStreamTableSource(Map) is default implemented now, and users may not override it.

indicate users to implement createStreamTableSource(context)?

wuchong · 2020-02-17T06:20:38Z

...api-java-bridge/src/main/java/org/apache/flink/table/factories/StreamTableSourceFactory.java

+	 * @param context context of this table source.
+	 * @return the configured table source.
+	 */
+	default StreamTableSource<T> createStreamTableSource(Context context) {


Do we need this interface?

Somebody implements StreamTableSourceFactory looks like need this method.

But he can still use the TableSource<T> createTableSource(context) interface.

if user use TableSource, there is no need to use StreamTableSourceFactory..

Yes. StreamTableSourceFactory should be deprecated. It is confused why there is a StreamTableSourceFactory and TableSourceFactory. But for compatibility, users can still use StreamTableSourceFactory but indicate users to use TableSource<T> createTableSource(context).

OK, I will remove this method.

...er-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/TableFactoryTest.scala

...blink/src/test/scala/org/apache/flink/table/planner/plan/utils/TestContextTableFactory.scala

...link-table-planner/src/test/scala/org/apache/flink/table/utils/TestContextTableFactory.scala

…actory

…ql-cli

…y context in blink planner

…ext in legacy planner

…ive connector

wuchong

LGTM.

Will merge it after tests passed in my branch.

JingsongLi requested review from twalthr and wuchong February 10, 2020 08:01

rmetzger added the review=description? label Feb 10, 2020

rmetzger added the component=TableSQL/API label Feb 10, 2020

wuchong reviewed Feb 11, 2020

View reviewed changes

JingsongLi force-pushed the tablefactory branch from 18beb9a to a9125de Compare February 11, 2020 09:22

JingsongLi force-pushed the tablefactory branch 5 times, most recently from 8c7c673 to b1f4018 Compare February 15, 2020 13:38

wuchong reviewed Feb 17, 2020

View reviewed changes

JingsongLi force-pushed the tablefactory branch from b1f4018 to 3428261 Compare February 18, 2020 05:48

[FLINK-15912][table] Add Context to TableSourceFactory and TableSinkF…

cf9cc83

…actory

JingsongLi force-pushed the tablefactory branch 2 times, most recently from 2db8738 to daf7fda Compare February 18, 2020 13:49

JingsongLi added 5 commits February 19, 2020 12:33

[FLINK-15912][table] Support create table source/sink by context in s…

e877210

…ql-cli

[FLINK-15912][table-planner-blink] Support create table source/sink b…

8171ac3

…y context in blink planner

[FLINK-15912][table-planner] Support create table source/sink by cont…

0d21b73

…ext in legacy planner

[FLINK-15912][table] Support create table source/sink by context in h…

4d4bc72

…ive connector

[FLINK-15912][table] Clean TableFactoryUtil

eba92d7

JingsongLi force-pushed the tablefactory branch from daf7fda to eba92d7 Compare February 19, 2020 04:33

wuchong approved these changes Feb 21, 2020

View reviewed changes

wuchong closed this in f6895da Feb 22, 2020

JingsongLi deleted the tablefactory branch April 26, 2020 05:32

[FLINK-15912][table] Add Context to TableSourceFactory and TableSinkFactory #11047

[FLINK-15912][table] Add Context to TableSourceFactory and TableSinkFactory #11047

Uh oh!

Conversation

JingsongLi commented Feb 10, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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

Uh oh!

flinkbot commented Feb 10, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Feb 10, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JingsongLi commented Feb 11, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

JingsongLi commented Feb 10, 2020 •

edited

Loading

flinkbot commented Feb 10, 2020 •

edited

Loading

JingsongLi commented Feb 11, 2020 •

edited

Loading