[FLINK-12604][table-api][table-planner] Register TableSource/Sink as CatalogTables #8549

dawidwys · 2019-05-27T06:59:39Z

What is the purpose of the change

This is next step in decoupling TableEnvironment from Calcite. It introduces registration of TableSource/Sink as CatalogTables.

Brief change log

This is based on #8521.

added org.apache.flink.table.catalog.ConnectorCatalogTable that wraps TableSource/Sink and used it for registration in TableEnvironment
added org.apache.flink.table.operations.TableSourceTableOperation for reading from an inline TableSource. This is used only when creating a Table from TableEnvironment#fromTableSource
removed unnecessary code duplication from classes such as TableSourceSinkTable, TableSourceTable, TableSinkTable etc.

Verifying this change

This change is already covered by existing tests.
Added test to verify TableEnviroment#fromTableSource behavior: org.apache.flink.table.runtime.stream.table.TableSourceITCase#testInlineCsvTableSource

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? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2019-05-27T07:01: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.

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

TableOperations. We do not store the DataStream & DataSet as Calcite's Tables anymore. We treat them as inline operations. When converting from TableOperations to RelNodes we directly create a special kind of DataStream/SetScan that does not access the catalog.

twalthr

+1 for 7192f70

twalthr

Awesome work @dawidwys. The code reads like a novel :D

I only had minor comments for 1de257b.

twalthr · 2019-06-03T11:36:00Z

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

 	 * <ol>
 	 *     <li>{@code [current-catalog].[current-database].[tablePath]}</li>
 	 *     <li>{@code [current-catalog].[tablePath]}</li>
 	 *     <li>{@code [tablePath]}</li>
 	 * </ol>
 	 *
 	 * @param tablePath table path to look for
-	 * @return {@link CatalogTableOperation} containing both fully qualified table identifier and its
-	 * {@link TableSchema}.
+	 * @return {@link ResolvedTable} wrapping original table with additional iformation about table path and


nit: information

twalthr · 2019-06-03T11:38:02Z

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

+	private final TableSink<T2> tableSink;
+	private final boolean isBatch;
+
+	private static final String COMMENT = "A table sink or source backed table.";


I meant "to back" that this Table is backed by a table sink or sourced.

Shall I maybe just remove the comment, the same way as we discussed here: #8521 (comment)

twalthr · 2019-06-03T11:40:41Z

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

+	}
+
+	@Override
+	public CatalogBaseTable copy() {


This violates the contract of the method. Is it actually required?

Unfortunately it is. All methods in the Catalog (add, get, alter etc.) call this method. I also don't like it.

twalthr · 2019-06-03T11:45:12Z

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

+						tableSource,
+						!connectorTable.isBatch(),
+						FlinkStatistic.UNKNOWN()))
+					.orElseThrow(() -> new TableException("Querying sink only table unsupported."));


rephrase: "Catalog table does only support sink operations."?

twalthr · 2019-06-03T11:56:35Z

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

@@ -425,20 +424,128 @@ abstract class TableEnvImpl(
        "Only tables that belong to this TableEnvironment can be registered.")
    }

-    checkValidTableName(name)


Should we add least at some checks for non-empty strings and not only whitespaces? Right now registerTableInternal fails in the check arguments of ObjectPath.

twalthr · 2019-06-03T12:04:32Z

...table-planner/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala

    val selectedFields: Option[Array[Int]])
  extends TableScan(cluster, traitSet, table) {

+


nit: empty line

twalthr · 2019-06-03T12:15:34Z

...ner/src/test/java/org/apache/flink/table/runtime/batch/table/JavaTableEnvironmentITCase.java

@@ -156,17 +156,6 @@ public void testTableRegister() throws Exception {
 		compareResultAsText(results, expected);
 	}

-	@Test(expected = TableException.class)
-	public void testIllegalName() throws Exception {


As mentioned earlier, let's keep this test but modify it with other illegal names.

twalthr · 2019-06-03T12:17:27Z

...e/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java

+				FlinkStatistic.UNKNOWN());
+
+			CatalogReader catalogReader = (CatalogReader) relBuilder.getRelOptSchema();
+			String refId = Integer.toString(System.identityHashCode(tableSourceTable.getTableSource()));


Can you add an explanation what we are doing here? Btw should we only use the refId or should we prefix it for readability unregistered_456789?

bowenli86 · 2019-06-03T18:18:32Z

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

+/**
+ * A {@link CatalogTable} that wraps a {@link TableSource} and/or {@link TableSink}.
+ * This allows registering those in a {@link Catalog}. It can not be persisted as the
+ * source and/or sink might be inline implementations and not be representable in a


a bit confused be the "might be" here.

Are some ConnectorCatalogTables inline impl, and some are not and can be converted to properties? The exception thrown in toProperties() indicates that all ConnectorCatalogTable cannot be converted to properties

It says the TableSource might be a inline implementation. You can either use TableEnvironment#connect than theoretically the table source is property serializable. The other possibility is just to use TableSource explicitly: TableEnvironment#fromTableSource.

The ConnectorCatalogTable therefore is always inline as we don't know which case we are handling.

bowenli86 · 2019-06-03T18:22:20Z

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

+public class ConnectorCatalogTable<T1, T2> extends AbstractCatalogTable {
+	private final TableSource<T1> tableSource;
+	private final TableSink<T2> tableSink;
+	private final boolean isBatch;


just want to double check: unlike a persistent catalog table that can be both batch and streaming, ConnectorCatalogTable can only be either batch or streaming, right?

In the long-term this property will not be necessary anymore. However, it is required in how the table environments (batch and streaming ones) work.

xuefuz · 2019-06-03T23:07:41Z

...e-api-java-bridge/src/main/java/org/apache/flink/table/operations/DataSetTableOperation.java

+ * {@link TableSchema} to the {@link org.apache.flink.api.common.typeutils.CompositeType}.
+ */
+@Internal
+public class DataSetTableOperation<E> extends TableOperation {


javadoc for ?

xuefuz · 2019-06-03T23:09:48Z

...e-api-java-bridge/src/main/java/org/apache/flink/table/operations/DataSetTableOperation.java

+@Internal
+public class DataSetTableOperation<E> extends TableOperation {
+
+	private final DataSet<E> dataStream;


Is there a special reason that we name the variable as dataStream instead of dataSet?

xuefuz · 2019-06-03T23:11:32Z

...pi-java-bridge/src/main/java/org/apache/flink/table/operations/DataStreamTableOperation.java

+ * {@link TableSchema} to the {@link org.apache.flink.api.common.typeutils.CompositeType}.
+ */
+@Internal
+public class DataStreamTableOperation<E> extends TableOperation {


javadoc for ?

…ables

twalthr

Thanks for the update @dawidwys. +1 from my side.

dawidwys requested a review from twalthr May 27, 2019 06:59

rmetzger added review=description? component=TableSQL/API component=TableSQL/LegacyPlanner labels May 27, 2019

dawidwys added 4 commits May 31, 2019 13:47

Tests adjusted - to be squashed before merge

da5ace9

Comments addressed

37ca357

Comments addressed 2

16b006d

dawidwys force-pushed the flink12604 branch from 0a6807f to 1de257b Compare June 3, 2019 10:27

twalthr approved these changes Jun 3, 2019

View reviewed changes

dawidwys force-pushed the flink12604 branch from 1de257b to b0b78fd Compare June 3, 2019 13:20

bowenli86 reviewed Jun 3, 2019

View reviewed changes

xuefuz reviewed Jun 3, 2019

View reviewed changes

dawidwys added 4 commits June 4, 2019 07:41

Unified output format of the description of TableOperation

e7183e1

[hotfix][table-api] Port TableSource specific descriptors

18acd13

[hotfix][table-planner] Fix getting external catalog

6d8a3c2

[FLINK-12604][table] Register a TableSource/Sink directly as CatalogT…

ccb9fa2

…ables

dawidwys force-pushed the flink12604 branch from b0b78fd to b63a53b Compare June 4, 2019 05:50

Comments addressed

1cdc6aa

dawidwys force-pushed the flink12604 branch from b63a53b to 1cdc6aa Compare June 4, 2019 06:39

twalthr approved these changes Jun 4, 2019

View reviewed changes

dawidwys closed this in 88c7d82 Jun 4, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12604][table-api][table-planner] Register TableSource/Sink as CatalogTables #8549

[FLINK-12604][table-api][table-planner] Register TableSource/Sink as CatalogTables #8549

dawidwys commented May 27, 2019

flinkbot commented May 27, 2019

twalthr left a comment

twalthr left a comment

twalthr Jun 3, 2019

twalthr Jun 3, 2019

dawidwys Jun 3, 2019

twalthr Jun 3, 2019

dawidwys Jun 3, 2019

twalthr Jun 3, 2019

twalthr Jun 3, 2019

twalthr Jun 3, 2019

twalthr Jun 3, 2019

twalthr Jun 3, 2019

bowenli86 Jun 3, 2019

dawidwys Jun 4, 2019

bowenli86 Jun 3, 2019

twalthr Jun 4, 2019

xuefuz Jun 3, 2019

xuefuz Jun 3, 2019

xuefuz Jun 3, 2019

twalthr left a comment

		val selectedFields: Option[Array[Int]])
		extends TableScan(cluster, traitSet, table) {

[FLINK-12604][table-api][table-planner] Register TableSource/Sink as CatalogTables #8549

[FLINK-12604][table-api][table-planner] Register TableSource/Sink as CatalogTables #8549

Conversation

dawidwys commented May 27, 2019

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 May 27, 2019

Review Progress

twalthr left a 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

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

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