[FLINK-25174][table] Introduce managed table interfaces and callback #18088

JingsongLi · 2021-12-13T07:59:44Z

What is the purpose of the change

Introduce managed table which introduced in https://cwiki.apache.org/confluence/display/FLINK/FLIP-188%3A+Introduce+Built-in+Dynamic+Table+Storage

Brief change log

4565223 Introduce CatalogBaseTable.TableKind.MANAGED and serialization for CatalogTable
e9f5cdc Introduce ManagedTableFactory interface
5bce725 Introduce Catalog.supportsManagedTable
8af06de Introduce TableDescriptor.forConnector.
801da05 Implement ManagedTableListener for managed table and callback in CatalogManager.

Verifying this change

CatalogBaseTableResolutionTest.testManagedPropertyDeSerialization
TableEnvironmentTest.testCreateManagedTableFromDescriptor

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, Kubernetes/Yarn, 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 / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2021-12-13T08:02:58Z

CI report:

075b187 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flinkbot · 2021-12-13T08:05:18Z

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 801da05 (Mon Dec 13 08:05:17 UTC 2021)

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.

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

Airblader · 2021-12-13T09:20:28Z

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

+        }
+        CatalogBaseTable origin = table.getOrigin();
+        // only DefaultCatalogTable or CatalogTableImpl is managed table
+        return origin instanceof DefaultCatalogTable || origin instanceof CatalogTableImpl;


What's the motivation for this restriction? How does a catalog implementor learn about this?

You are right, we should exclude InlineCatalogTable and ConnectorCatalogTable these two troublemakers.

For InlineCatalogTable, it is a little tricky like:

try { origin.getOptions(); } catch (TableException ignore) { // exclude abnormal tables, such as InlineCatalogTable that does not have the options return false; }

Airblader · 2021-12-13T09:21:56Z

...able/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java

+        if (ignoreIfExists) {
+            tEnv.executeInternal(createOperation);
+        } else {
+            try {


This entire try-catch block could be written in a nicer way using AssertJ's assertThatThrownBy.

Airblader · 2021-12-13T09:23:11Z

...able/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java

@@ -110,4 +118,96 @@ public void testTableFromDescriptor() {

        assertEquals("fake", lookupResult.get().getTable().getOptions().get("connector"));
    }
+
+    @Test
+    public void testCreateManagedTableFromDescriptor() {


I would prefer splitting this into separate test cases. If one case fails, we'll immediately know which one. Right now this is one giant test case that actually tests several cases, and several things per case.

...e/flink-table-api-java/src/test/java/org/apache/flink/table/api/TestManagedTableFactory.java

Airblader · 2021-12-13T09:32:38Z

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

+import static org.apache.flink.table.factories.ManagedTableFactory.discoverManagedTableFactory;
+
+/** The listener for managed table operations. */
+public class ManagedTableListener {


Please add an API stability annotation here.

Airblader · 2021-12-13T09:35:18Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java

@@ -311,6 +311,14 @@ void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfE
    void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
            throws TableNotExistException, CatalogException;

+    /**
+     * If return true, the Table without specified connector will be translated to the Flink managed
+     * table. See {@link CatalogBaseTable.TableKind#MANAGED}


Use @see?

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java

Airblader · 2021-12-13T09:37:08Z

...-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogBaseTable.java

@@ -39,6 +39,7 @@
    @PublicEvolving
    enum TableKind {
        TABLE,
+        MANAGED,


It'd be nice to add some JavaDoc for this value what a managed table is.

Airblader · 2021-12-13T09:46:13Z

...ble/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java

        this.schema = checkNotNull(schema, "Schema must not be null.");
        this.comment = comment;
        this.partitionKeys = checkNotNull(partitionKeys, "Partition keys must not be null.");
        this.options = checkNotNull(options, "Options must not be null.");
+        this.tableKind = checkNotNull(tableKind, "Table kind must not be null.");


We should validate that it isn't VIEW either. In fact, would it maybe make sense to introduce a separate DefaultManagedTable (which maybe inherits from DefaultCatalogTable) instead of adding the table kind here?

We can introduce a DefaultManagedTable.

…table/catalog/Catalog.java Co-authored-by: Ingo Bürk <admin@airblader.de>

…k/table/api/TestManagedTableFactory.java Co-authored-by: Ingo Bürk <admin@airblader.de>

JingsongLi · 2021-12-14T06:06:48Z

@flinkbot run azure

twalthr

Thanks for the PR @JingsongLi. Maybe it was not a good idea to introduce an additional table kind for managed tables. From the changes, I don't think it actually provides much benefit. A utility method that simply check the getOptions() of a CatalogTable could also do the job. It would reduce the number of changes in this PR to a minimum. What do you think?

In the end the only thing that matters is whether connector= is present or not. We can neglect the legacy.

twalthr · 2021-12-14T15:23:17Z

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

+                                        getCatalog(objectIdentifier.getCatalogName()).orElse(null),
+                                        objectIdentifier,
+                                        resolveCatalogBaseTable(table),
+                                        true,


aren't the properties "temporary" and "catalog = null" related? I don't think we need the temporary flag. we can still add it on the internal API if necessary.

Our temporary tables can be catalog-owned, I remember this is why we do: https://issues.apache.org/jira/browse/FLINK-21185

FYI: We also do org.apache.flink.table.catalog.CatalogManager.TableLookupResult#isTemporary. So the temporary table might be catalog owned during creation but will look like a non-temporary one currently in the factory.

But I'm fine with the change.

I got it. Here is for temporary managed table.

twalthr · 2021-12-14T15:24:28Z

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

+                            managedTableListener.notifyTableCreation(
+                                    catalog,
+                                    objectIdentifier,
+                                    resolveCatalogBaseTable(table),


nit: introduce a local variable here and below, it avoids too much nesting and makes it easier to set breakpoints during debugging

twalthr · 2021-12-14T15:32:27Z

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

+            return false;
+        }
+
+        if (!StringUtils.isNullOrWhitespaceOnly(


we call table.getOptions() very often here. how about calling it once in a try-catch and than just check for the remaining cases?

twalthr · 2021-12-14T15:44:17Z

...e/flink-table-api-java/src/test/java/org/apache/flink/table/api/TestManagedTableFactory.java

+                (k, v) -> {
+                    if (v != null) {
+                        if (v.get() == null) {
+                            v.set(context.getCatalogTable().toProperties());


isn't calling getOptions() enough, do we really need to serialize the schema as well?

twalthr · 2021-12-14T15:46:33Z

...e/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java

@@ -79,6 +80,10 @@
                properties.put(COMMENT, comment);
            }

+            if (resolvedTable.getTableKind() == TableKind.MANAGED) {


why do we need this? can't we derive the "managed" property from the options? looking for a missing connector should already be enough? I think we can assume that connector.type is not used by now.

twalthr · 2021-12-14T15:49:28Z

...ble/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultManagedTable.java

+
+/** Default implementation of a managed {@link CatalogTable}. */
+@Internal
+public class DefaultManagedTable extends DefaultCatalogTable {


let's drop this class and enrich DefaultCatalogTable with an additional constructor argument. it is internal anyway and implementers should not do instanceof checks against this class.

twalthr · 2021-12-14T15:53:26Z

...able-planner/src/main/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchema.java

@@ -86,6 +86,7 @@ private FlinkStatistic getStatistic(
        final ResolvedCatalogBaseTable<?> resolvedBaseTable = lookupResult.getResolvedTable();
        switch (resolvedBaseTable.getTableKind()) {
            case TABLE:
+            case MANAGED:


I thought about my previous comment around a special TableKind again. I fear we are mixing metadata encoded in table kind and metadata encoded in options.

JingsongLi · 2021-12-15T02:58:41Z

Thanks for the PR @JingsongLi. Maybe it was not a good idea to introduce an additional table kind for managed tables. From the changes, I don't think it actually provides much benefit. A utility method that simply check the getOptions() of a CatalogTable could also do the job. It would reduce the number of changes in this PR to a minimum. What do you think?

In the end the only thing that matters is whether connector= is present or not. We can neglect the legacy.

+1, The biggest benefit is the distinction between table and managed, but as the implementation looks at:

planner identifies managed table by just looking at the catalog and options
catalog identifies managed table by just looking at the options

Yes, the benefits are not big

slinkydeveloper · 2021-12-15T08:09:00Z

...le-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java

+        assertThat(resolvedTable.toProperties(), equalTo(properties));
+
+        assertThat(resolvedTable.getResolvedSchema(), equalTo(RESOLVED_TABLE_SCHEMA));


Please use only assertj assertions, and convert the tests here to use assertj

slinkydeveloper · 2021-12-15T08:30:02Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java

+     *
+     * @see CatalogBaseTable.TableKind#MANAGED
+     */
+    default boolean supportsManagedTable() {


I might miss a point in the original design, but why the catalog needs to support managed tables? Isn't the support already provided by CatalogManager itself?

There are some catalogs that already use table without connector as their catalog internal table.

You can see the first few replies of the FLIP discussion.

slinkydeveloper · 2021-12-15T08:31:26Z

...able/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java

+        if (ignoreIfExists) {
+            tEnv.executeInternal(createOperation);
+        } else {
+            Assertions.assertThrows(


Please use assertj for testing

slinkydeveloper · 2021-12-15T08:36:21Z

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

-                    String.format(
-                            "Table options do not contain an option key '%s' for discovering a connector.",
-                            CONNECTOR.key()));
+            ManagedTableFactory factory =


More than having an empty connector id and hardcoding the discovery of ManagedTableFactory, I wonder if we can have a connector id for that, and then we inject it from TableEnvironmentImpl every time we parse a table which doesn't provide a connector value in the table options.

I prefer the empty string to other ids because the empty string itself is a special symbol that is often mixed up with "none".

tbh I also don't like the empty string. A managed-table or even default (which we use for planner and executor discovery) would be nicer and looks less like a bug. default would ensure consistency.

You have convinced me that I think we can use default

twalthr

Thanks for the update @JingsongLi. The changes look much better now. I just had a last comment on the user experience in case of errors.

twalthr · 2021-12-15T17:24:58Z

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

+            if (!factoryClass.isAssignableFrom(factory.getClass())) {
+                throw new ValidationException(
+                        String.format(
+                                "The managed table factory '%s' dose not implement '%s'.",


the exception contains a typo and is not very nice. in the end we should have a helpful exception that includes the old exception:

Table options do not contain an option key '%s' for discovering a connector. Therefore, Flink assumes a managed table. However, a managed table factory is not in the classpath.

Either we wrap the original exception or call some internal discoverFactory with an optional return type.

Nice! I prefer to throw new exception directly, we don't need repetitive old exception.

I created FactoryUtil.discoverManagedTableFactory, it throws exception messages for not found and multiple found.

JingsongLi · 2021-12-16T09:00:41Z

@flinkbot run azure

JingsongLi · 2021-12-16T09:09:26Z

Hi @twalthr @slinkydeveloper @Airblader Can you see if there are any other questions? (Or maybe I missed something)

twalthr

LGTM

JingsongLi · 2021-12-17T08:03:03Z

Thanks for your review! Merged.

This closes apache#18088

JingsongLi added 5 commits December 13, 2021 14:50

[FLINK-25174][table-api] Introduce CatalogBaseTable.TableKind.MANAGED

4565223

[FLINK-25174][table] Introduce ManagedTableFactory

e9f5cdc

[FLINK-25174][table-api] Introduce Catalog.supportsManagedTable

5bce725

[FLINK-25175][table-api] Introduce TableDescriptor.forConnector

8af06de

[FLINK-25174][table] Implement callback for managed table

801da05

rmetzger added the component=TableSQL/API label Dec 13, 2021

Airblader reviewed Dec 13, 2021

View reviewed changes

JingsongLi and others added 3 commits December 14, 2021 11:01

Update flink-table/flink-table-common/src/main/java/org/apache/flink/…

43c8f02

…table/catalog/Catalog.java Co-authored-by: Ingo Bürk <admin@airblader.de>

Update flink-table/flink-table-api-java/src/test/java/org/apache/flin…

0cff4b3

…k/table/api/TestManagedTableFactory.java Co-authored-by: Ingo Bürk <admin@airblader.de>

Address comments

8b16953

JingsongLi added 2 commits December 14, 2021 14:14

Remove exclution

3453bb8

Fix cases

ebf93b8

twalthr requested changes Dec 14, 2021

View reviewed changes

JingsongLi added 3 commits December 15, 2021 11:22

Address comments

5288bce

Fix compile

7660ffb

Fix FactoryUtilTest

2d0504a

slinkydeveloper reviewed Dec 15, 2021

View reviewed changes

JingsongLi added 3 commits December 15, 2021 17:27

Filter empty Identifier in exception

f928bf4

Use assertj only

d8ce847

Fix TableEnvironmentTest.testAlterTableResetRequiredOptionKey

f5d5ba0

twalthr requested changes Dec 15, 2021

View reviewed changes

Improve error message

08822a7

twalthr approved these changes Dec 16, 2021

View reviewed changes

JingsongLi added 2 commits December 17, 2021 10:16

Use default factoryIdentifier

3bfb0e8

Fix case

075b187

JingsongLi closed this in f01e911 Dec 17, 2021

niklassemmler pushed a commit to niklassemmler/flink that referenced this pull request Feb 3, 2022

[FLINK-25174][table] Implement callback for managed table

e295b4f

This closes apache#18088

JingsongLi deleted the managed_interface branch March 8, 2022 07:12

		assertThat(resolvedTable.toProperties(), equalTo(properties));

		assertThat(resolvedTable.getResolvedSchema(), equalTo(RESOLVED_TABLE_SCHEMA));

[FLINK-25174][table] Introduce managed table interfaces and callback #18088

[FLINK-25174][table] Introduce managed table interfaces and callback #18088

Conversation

JingsongLi commented Dec 13, 2021 • 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

flinkbot commented Dec 13, 2021 • edited Loading

CI report:

flinkbot commented Dec 13, 2021

Automated Checks

Review Progress

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JingsongLi Dec 14, 2021 • 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

JingsongLi commented Dec 14, 2021

twalthr 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

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

JingsongLi commented Dec 15, 2021

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

twalthr 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

JingsongLi commented Dec 16, 2021

JingsongLi commented Dec 16, 2021

twalthr left a comment

Choose a reason for hiding this comment

JingsongLi commented Dec 17, 2021

JingsongLi commented Dec 13, 2021 •

edited

Loading

flinkbot commented Dec 13, 2021 •

edited

Loading

JingsongLi Dec 14, 2021 •

edited

Loading