[FLINK-14490][table] Add methods for interacting with temporary objects #9971

dawidwys · 2019-10-22T16:20:48Z

What is the purpose of the change

This PR adds methods for interacting with temporary objects as described in FLIP-64:
https://cwiki.apache.org/confluence/display/FLINK/FLIP-64%3A+Support+for+Temporary+Objects+in+Table+module

Verifying this change

This change added tests.

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 / no)
If yes, how is the feature documented? to be done separately (see https://issues.apache.org/jira/browse/FLINK-14486)

flinkbot · 2019-10-22T16:23:59Z

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 87124ec (Wed Dec 04 14:50:47 UTC 2019)

Warnings:

2 pom.xml files were touched: Check for build and licensing issues.
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

flinkbot · 2019-10-22T16:55:18Z

CI report:

754bf46 : SUCCESS Build
c23e258 : FAILURE Build
ce946ec : SUCCESS Build
0fbaa98 : SUCCESS Build
70eb2b6 : SUCCESS Build
bb9775f : FAILURE Build
87124ec : SUCCESS Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

twalthr

+1 for 3424926

twalthr

Some feedback for eab35b0.

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/calcite/CalciteParser.scala

...-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala

twalthr · 2019-10-23T14:27:38Z

...-table-planner/src/test/java/org/apache/flink/table/sqlexec/SqlToOperationConverterTest.java

 			"    'connector' = 'kafka', \n" +
 			"    'kafka.topic' = 'log.test'\n" +
 			")\n";
 		final FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT);


Is the planner by dialect still necessary if we also create a parser by dialect?

It uses the dialect to create a parser for view expansion. (though we do not support views expansion yet).

twalthr

Some comments for b4030de

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

twalthr

Minor comments for 4e9ee4e.

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Parser.java

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Planner.java

...rc/test/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImplTest.scala

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/ParserImpl.scala

twalthr · 2019-10-23T14:52:45Z

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/ParserImpl.scala

+
+class ParserImpl(
+    catalogManager: CatalogManager,
+    validatorProvider: () => FlinkPlannerImpl,


plannerProvider

In the context of the parser it serves only as a validator. We do not use the toRel conversion part. Ideally it would be best to further split the FlinkPlanner class into subclasses, but did not have enough time.

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/ParserImpl.scala

twalthr

Thanks @dawidwys. 4122516 looks great I added rather minor comments.

twalthr · 2019-10-23T15:11:15Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

 					// wrapper contains only sink (not source)
 					ConnectorCatalogTable sourceAndSink = ConnectorCatalogTable
 						.sourceAndSink(sourceSinkTable.getTableSource().get(), tableSink, !IS_STREAM_TABLE);
-					catalogManager.alterTable(sourceAndSink, getTemporaryObjectIdentifier(name), false);


alterTable is not used anymore, right?

Correct, alterTable works with permanent table. We do not have an API for that.

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

...planner-blink/src/main/java/org/apache/flink/table/planner/catalog/CatalogCalciteSchema.java

...-blink/src/main/java/org/apache/flink/table/planner/catalog/CatalogManagerCalciteSchema.java

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

twalthr · 2019-10-23T15:31:09Z

...lanner-blink/src/test/java/org/apache/flink/table/planner/catalog/CatalogStatisticsTest.java

 		Catalog catalog = tEnv.getCatalog(tEnv.getCurrentCatalog()).orElse(null);
 		assertNotNull(catalog);
+		catalog.createTable(
+			ObjectPath.fromString("default_database.T1"),


side comment: we should get rid of this method, it doesn't even support the identifier escaping

twalthr

We should improve the docs a bit further in 7d6721c.

...ble-api-java-bridge/src/main/java/org/apache/flink/table/api/java/BatchTableEnvironment.java

...le-api-java-bridge/src/main/java/org/apache/flink/table/api/java/StreamTableEnvironment.java

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

...api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/BatchTableEnvironment.scala

twalthr

One little comment for 90bf886

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

twalthr

some suggestions for 64d8ca6

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java

twalthr

Some feedback for 754bf46.

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java

...-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/CalciteParser.scala

...nk-table-api-java/src/main/java/org/apache/flink/table/descriptors/BatchTableDescriptor.java

...-table-api-java/src/main/java/org/apache/flink/table/descriptors/ConnectTableDescriptor.java

twalthr · 2019-10-23T16:42:10Z

Thanks for this important API refactoring. The code changes look mostly good but we should definitely further improve the JavaDocs to avoid any confusion for our existing and future users. We should also update the website documentation and add good release notes to JIRA when closing this issue. This change will annoy existing users. We need to communicate the reasons for these changes.

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

...e/flink-table-planner/src/main/java/org/apache/flink/table/catalog/CatalogCalciteSchema.java

dawidwys · 2019-10-24T14:55:25Z

Thank you @bowenli86 @twalthr @zjuwangg for the reviews.

I did first pass over your comments. Still need to apply a few remaining comments.

dawidwys · 2019-10-25T14:02:45Z

Hey @bowenli86 could you have another pass over the PR. I think @twalthr will not be able to do it any time soon. I tried to address all of his comments as well as yours and @zjuwangg.

bowenli86 · 2019-10-25T20:34:51Z

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

+	private Optional<TableLookupResult> getPermanentTable(ObjectIdentifier objectIdentifier)
+			throws TableNotExistException {
+		Catalog currentCatalog = catalogs.get(objectIdentifier.getCatalogName());
+		ObjectPath objectPath = new ObjectPath(


this call can be saved by calling objectIdentifier.toObjectPath()

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

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

bowenli86 · 2019-10-25T20:48:38Z

...ble-api-java-bridge/src/main/java/org/apache/flink/table/api/java/BatchTableEnvironment.java

+	 * <p>The field names of the {@link Table} are automatically derived
+	 * from the type of the {@link DataSet}.
+	 *
+	 * <p>The view is registered in the current catalog and database. To register the view in


this comment can be a bit misleading as the view is registered actually only in the namespace of current catalog and db, but isn't physically located in the catalog/db, right?

Will change to The view is registered in the namespace of the current catalog and database.

bowenli86 · 2019-10-25T20:51:03Z

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

+		return doDropTemporaryTable(identifier, (table) -> table instanceof CatalogView);
+	}
+
+	private boolean doDropTemporaryTable(


the API naming of "doXxx()" feels a bit weird. rename to dropTemporaryTableInternal()?

bowenli86 · 2019-10-25T20:53:04Z

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

 	 *                          does not exist.
 	 */
 	public void dropTable(ObjectIdentifier objectIdentifier, boolean ignoreIfNotExists) {
+		if (temporaryTables.containsKey(objectIdentifier)) {


this behavior feels a bit abnormal to me. Why cannot users drop a catalog table when a temp table with same name space exist?

Table and temp table creation and deletion should be independent of each other, e,g, from sql perspective, users should be able to run "CREATE/DROP TABLE" and "CREATE/DROP TEMP TABLE" independently

Just curious about this, I took a look to spark:

* If a database is specified in `name`, this will drop the table from that database. * If no database is specified, this will first attempt to drop a temporary view with * the same name, then, if that does not exist, drop the table from the current database.

Curious about other databases, I don't know how they behave.

Spark does not have 3-part qualified temporary objects. We followed MySQL on that topic.

This behavior is introduced to prevent unintentional drops of temporary tables, which might have very costly undesirable effects. Users are usually accustomed with DROP TABLE, but not necessarily with DROP TEMPORARY TABLE. If a user issues a DROP TABLE with the intent of dropping the temporary table it will actually drop in a drop of the permanent one.

Dropping tables is a very invasive operation. I tried to design the behaviour in a way to force the user make a really conscious decision on which table should be dropped.

BTW, as a side note, this behaviour is described in the FLIP-64. There were no objections towards that behaviour during the discussion.

bowenli86 · 2019-10-25T20:58:26Z

Hi Dawid, I was only able to review the last five commits as I don't really know much about planner.

This PR, as well as several individual commits, is quite huge, and it makes really hard to review and follow. Would be nice to break such good amount of work down into several smaller PRs next time.

would be good to have your reviews if you have time. @KurtYoung @JingsongLi @xuefuz

KurtYoung

Some minor comments for 988bd85

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

...ble-api-java/src/main/java/org/apache/flink/table/operations/CatalogSinkModifyOperation.java

...table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateTableOperation.java

...k-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/DropTableOperation.java

KurtYoung

one question for b426a97

KurtYoung · 2019-10-28T03:40:56Z

...e-planner-blink/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala

-        case e: CSqlParseException =>
-          throw new SqlParserException(s"SQL parse failed. ${e.getMessage}", e)
-      }
+      val sqlNode: SqlNode = parser.parse(queryString)


Is it intended to re-use parser here? I noticed in PlannerBase::parser you pointed out that you didn't want to reuse parser since config might changed.

Yes, because the whole FlinkPlannerImpl is used only once per planning session.

JingsongLi

Thanks dawid, left some comments.

JingsongLi · 2019-10-28T03:26:59Z

...le/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala

-          .toJava(scanInternal(Array(name)).map(t => new TableReferenceExpression(name, t)))
+          .toJava(
+            Try({
+              val unresolvedIdentifier = UnresolvedIdentifier.of(name)


Why use Try? If it is an illegal name, should we throw exception?

This gives positive results when resolving IN expression. But will be invoked for all UnresolvedExpressions. That's why it might not necessary be a table identifier. The contract of the ExpressionResolver is that it actually never throws exception.

As a side note. The way TableIdentifier is currently implemented is actually wrong. This method is primarily for resolving SELECT * FROM t WHERE f IN (tableName), which is actually incorrect SQL statement. It is not allowed to use identifiers in an IN clause. Therefore once we introduce the Java Expression DSL we should get rid of this catalog lookup for a TableReference.

Got it. You can put these comments to code.

JingsongLi · 2019-10-28T03:30:34Z

...le/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala

+    val unresolvedIdentifier = UnresolvedIdentifier.of(parser.parseIdentifier(path).names: _*)
+    scanInternal(unresolvedIdentifier) match {
+      case Some(table) => createTable(table)
+      case None => throw new TableException(s"Table '$path' was not found.")


Can we have a unify exception message? Use unresolvedIdentifier?

JingsongLi · 2019-10-28T03:38:49Z

...-table-api-java/src/main/java/org/apache/flink/table/descriptors/ConnectTableDescriptor.java

+		Map<String, String> properties = new HashMap<>(toProperties());
+		schemaProperties.keySet().forEach(properties::remove);
+
+		CatalogTableImpl catalogTable = new CatalogTableImpl(


CatalogTable has partitionKeys too, consider add partition keys in CatalogTableImpl.toProperties and parse partition keys here?
(We can add later too)

Yes I agree we should do that. I think we can do that in a follow-up? If I understand it correctly it will still use the ProjectableTableSource if we do not expose the partitions through the CatalogTable right?

WDYT?

OK.
After https://issues.apache.org/jira/browse/FLINK-14381 , we removed partition support for temporary table for clean up PartitionableTableSource/Sink.
I created JIRA to track it: https://issues.apache.org/jira/browse/FLINK-14543 , will fixed it in 1.10.

JingsongLi · 2019-10-28T03:39:18Z

...-table-api-java/src/main/java/org/apache/flink/table/descriptors/ConnectTableDescriptor.java

+		Map<String, String> properties = new HashMap<>(toProperties());
+		schemaProperties.keySet().forEach(properties::remove);
+
+		CatalogTableImpl catalogTable = new CatalogTableImpl(


Maybe we can use CatalogTableBuilder here?
Looks like there some bugs in CatalogTableBuilder, it not remove schemaProperties in properties?

We could use the CatalogTableBuilder, yes. Wasn't aware of the class. I am not sure though, if we need the CatalogTableBuilder class in a long run. I think the functionality is actually duplicated with the one in ConnectTableDescriptor.

JingsongLi · 2019-10-28T03:55:47Z

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

 	 *                          does not exist.
 	 */
 	public void dropTable(ObjectIdentifier objectIdentifier, boolean ignoreIfNotExists) {
+		if (temporaryTables.containsKey(objectIdentifier)) {


Just curious about this, I took a look to spark:

* If a database is specified in `name`, this will drop the table from that database. * If no database is specified, this will first attempt to drop a temporary view with * the same name, then, if that does not exist, drop the table from the current database.

Curious about other databases, I don't know how they behave.

KurtYoung

comment for 77880fc

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

KurtYoung

some comment for Parser's interface: dfc8f07

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Parser.java

dawidwys · 2019-10-28T09:05:22Z

I tried to address all your comments @KurtYoung @JingsongLi @bowenli86

Could you have another look?

KurtYoung

Thanks @dawidwys for this big effort, it indeed cleaned lots of APIs. The changes LGTM now and +1 to merge.

…face

dawidwys · 2019-10-29T12:59:49Z

Thanks for the reviews. Merging

dawidwys changed the title ~~[FLINK-14490] Add methods for interacting with temporary objects~~ [FLINK-14490][table] Add methods for interacting with temporary objects Oct 22, 2019

rmetzger added review=description? component=TableSQL/Planner component=TableSQL/API component=TableSQL/LegacyPlanner labels Oct 22, 2019

twalthr approved these changes Oct 23, 2019

View reviewed changes

twalthr requested changes Oct 23, 2019

View reviewed changes

...-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java Outdated Show resolved Hide resolved

twalthr requested changes Oct 23, 2019

View reviewed changes

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java Outdated Show resolved Hide resolved

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java Outdated Show resolved Hide resolved

twalthr requested changes Oct 23, 2019

View reviewed changes

bowenli86 reviewed Oct 23, 2019

View reviewed changes

JingsongLi mentioned this pull request Oct 24, 2019

[FLINK-14381][table] Partition field names should be got from CatalogTable instead of source/sink #9909

Closed

dawidwys force-pushed the flip-64-2 branch from 754bf46 to c23e258 Compare October 24, 2019 14:54

dawidwys force-pushed the flip-64-2 branch 2 times, most recently from ce946ec to 0fbaa98 Compare October 25, 2019 13:59

bowenli86 reviewed Oct 25, 2019

View reviewed changes

KurtYoung reviewed Oct 28, 2019

View reviewed changes

JingsongLi reviewed Oct 28, 2019

View reviewed changes

KurtYoung reviewed Oct 28, 2019

View reviewed changes

.../flink-table-api-java/src/main/java/org/apache/flink/table/catalog/UnresolvedIdentifier.java Show resolved Hide resolved

KurtYoung reviewed Oct 28, 2019

View reviewed changes

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Parser.java Show resolved Hide resolved

dawidwys force-pushed the flip-64-2 branch from 0fbaa98 to 70eb2b6 Compare October 28, 2019 09:04

KurtYoung approved these changes Oct 28, 2019

View reviewed changes

dawidwys added 8 commits October 29, 2019 09:15

[FLINK-14490] Use ObjectIdentifier in TableOperations

833dd1a

[FLINK-14490][table-planner] Extract CalciteParser from FlinkPlannerImpl

237281d

[FLINK-14490][table-api] Introduce UnresolvedIdentifier

3cc5ab4

[FLINK-14490][table-planner] Extract parsing logic from Planner inter…

e6b70fe

…face

[FLINK-14490][table] Introduced temporary tables to CatalogManager

095d3ed

[FLINK-14490][table-api] Add createTemporaryView

fc21839

[FLINK-14490][table-api] Add drop temporary tables

d4cee93

[FLINK-14490][table-api] Add from method

91e1a5e

dawidwys force-pushed the flip-64-2 branch from 70eb2b6 to bb9775f Compare October 29, 2019 08:16

[FLINK-14490][table-api] Rework insertInto method

87124ec

dawidwys force-pushed the flip-64-2 branch from bb9775f to 87124ec Compare October 29, 2019 10:28

dawidwys merged commit fe966d4 into apache:master Oct 29, 2019

[FLINK-14490][table] Add methods for interacting with temporary objects #9971

[FLINK-14490][table] Add methods for interacting with temporary objects #9971

Uh oh!

Conversation

dawidwys commented Oct 22, 2019

What is the purpose of the change

Verifying this change

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

Documentation

Uh oh!

flinkbot commented Oct 22, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Checks

Review Progress

Uh oh!

flinkbot commented Oct 22, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

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!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

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!

Uh oh!

twalthr left a comment

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

flinkbot commented Oct 22, 2019 •

edited

Loading

flinkbot commented Oct 22, 2019 •

edited

Loading

bowenli86 Oct 25, 2019 •

edited

Loading