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

JingsongLi · 2019-10-16T08:37:08Z

What is the purpose of the change

Now PartitionableTableSource and PartitionableTableSink have "getPartitionFieldNames" method, this should be removed, and planner rules should get it from CatalogManager.

The partition field names are the information of Table, source/sink should only be fed with such information but not get them out of it.

Brief change log

See commits.

Verifying this change

This change is already covered by existing tests.

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

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

Documentation

Does this pull request introduce a new feature? no

flinkbot · 2019-10-16T08:40:54Z

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 6a82d9f (Wed Dec 04 14:47:23 UTC 2019)

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

flinkbot · 2019-10-16T09:37:17Z

CI report:

f20e67a : SUCCESS Build
3c1480f : SUCCESS Build
e3ba6b1 : SUCCESS Build
e5c0586 : SUCCESS Build
4c59ed7 : SUCCESS Build
ca20ef1 : CANCELED Build
6a82d9f : SUCCESS Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

docete

Thanks for your PR. I leave some comments.
btw: the last 3 commit's have wrong commit msg (table-planner-planner -> table-planner-blink).

...lanner/src/test/scala/org/apache/flink/table/runtime/batch/sql/PartitionableSinkITCase.scala

...rc/test/scala/org/apache/flink/table/planner/runtime/batch/sql/PartitionableSinkITCase.scala

...table-planner-blink/src/main/scala/org/apache/flink/table/planner/sinks/TableSinkUtils.scala

...rc/test/scala/org/apache/flink/table/planner/runtime/batch/sql/PartitionableSinkITCase.scala

JingsongLi · 2019-10-24T03:35:33Z

Thanks for your PR. I leave some comments.
btw: the last 3 commit's have wrong commit msg (table-planner-planner -> table-planner-blink).

Thanks for your review, updated.

docete · 2019-10-24T07:06:06Z

LGTM now!

JingsongLi · 2019-10-24T07:40:41Z

@wuchong @KurtYoung Can you take a look?

KurtYoung · 2019-10-24T07:41:16Z

For LogicalSink, my feeling is we should provide enough information such as partition keys at first step instead of looking to catalog manager during conversion. Have you considered about this approach?

KurtYoung · 2019-10-24T07:41:45Z

In PlannerBase, we actually already know whether it's a PartitionableTableSink

JingsongLi · 2019-10-24T07:50:35Z

For LogicalSink, my feeling is we should provide enough information such as partition keys at first step instead of looking to catalog manager during conversion. Have you considered about this approach?

But consider the future needs, we need get some other informations from catalog table or catalog, we don't need to pass all informations to LogicalSink, for example, we need do the partition pruning from catalog, it is better to just pass an identifier.

KurtYoung · 2019-10-24T08:19:15Z

Consider a temporal partition table, I'm not sure such table would stored also in CatalogManager. So you have to write the logic like:
tryGetTableFromTemporalStorage(); if not temporal, get it from CatalogManger every time you want to access some information of the table.

Another solution is we can pass in the CatalogTable? It serves like a meta about the table you trying to read and write.

wuchong

My gut feeling is we should pass in all the required information on CatalogTable to LogicalSink,
other information we can access them via Catalog. I think this is also the way for source, i.e. watermark, computed column, partition.

...link-table-common/src/main/java/org/apache/flink/table/sources/PartitionableTableSource.java

JingsongLi · 2019-10-24T08:31:11Z

Consider a temporal partition table, I'm not sure such table would stored also in CatalogManager. So you have to write the logic like:
tryGetTableFromTemporalStorage(); if not temporal, get it from CatalogManger every time you want to access some information of the table.

Another solution is we can pass in the CatalogTable? It serves like a meta about the table you trying to read and write.

According to #9971 , temporal table is stored in CatalogManger, so we can get the informations from CatalogManger.

CatalogTable not contains identifier information. Catalog contains more information than it does, such as stats.

KurtYoung · 2019-10-24T08:33:30Z

If you want, you can pass both table identifier as well as CatalogTable. The key point is we want to reduce unnecessary access to catalog manager.

JingsongLi · 2019-10-24T08:35:15Z

Hi @KurtYoung @wuchong , can you explain more? Why we want to reduce unnecessary access to catalog manager?

KurtYoung · 2019-10-24T09:10:56Z

3 minor reasons, none of them are critical but kind of bothers me:

Accessing catalog manager may introduce external system access and could cost some time which will increase the optimization duration.
Access catalog manager multiple times could cause data inconsistency. We already get such information before entering to optimization phase, and information might changed when you look up the catalog manager again.
As pointed out in ForwardFields Optimizer integration #2, we already get all the information you needed before optimization, why bother to get it again?

JingsongLi · 2019-10-24T09:39:40Z

3 minor reasons, none of them are critical but kind of bothers me:

Accessing catalog manager may introduce external system access and could cost some time which will increase the optimization duration.

Access catalog manager multiple times could cause data inconsistency. We already get such information before entering to optimization phase, and information might changed when you look up the catalog manager again.

As pointed out in ForwardFields Optimizer integration #2, we already get all the information you needed before optimization, why bother to get it again?

Thanks @KurtYoung to explain it. none of them are critical but can convince me, I'll update it. But I think I can keep the pass of identifier and catalog manager, what do you think?

KurtYoung · 2019-10-24T09:52:04Z

You mean catalog table, not catalog manager, right?

3 minor reasons, none of them are critical but kind of bothers me:

Accessing catalog manager may introduce external system access and could cost some time which will increase the optimization duration.

Access catalog manager multiple times could cause data inconsistency. We already get such information before entering to optimization phase, and information might changed when you look up the catalog manager again.

As pointed out in ForwardFields Optimizer integration #2, we already get all the information you needed before optimization, why bother to get it again?

Thanks @KurtYoung to explain it. none of them are critical but can convince me, I'll update it. But I think I can keep the pass of identifier and catalog manager, what do you think?

You mean catalog table, not catalog manager, right?

JingsongLi · 2019-10-24T09:57:48Z

You mean catalog table, not catalog manager, right?

I mean pass identifier and catalog table to logical sink and logical source, and keep catalog manager in FlinkContext.

KurtYoung · 2019-10-24T10:04:05Z

I'm ok with table identifier and catalog table, but not introduce catalog manager to FlinkContext yet. You don't need it in this PR, right?

JingsongLi · 2019-10-24T10:11:49Z

I'm ok with table identifier and catalog table, but not introduce catalog manager to FlinkContext yet. You don't need it in this PR, right?

OK, I'll remove catalog manager and identifier.

KurtYoung

I think i might have found some logic flaws with the old design. Let me know whether it make sense to you.

KurtYoung · 2019-10-24T11:21:50Z

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

 					isStreamingMode,
-					FlinkStatistic.builder().tableStats(tableStats).build());
+					FlinkStatistic.builder().tableStats(tableStats).build(),
+					null);


Will we also have CatalogTable for such TableSource wrapped table in the future? If not, I would suggest to change CatalogTable to Opion[CatalogTable] in TableSourceTable.

Actually, I have some concern to put the whole CatalogTable in TableSourceTable. Because in CatalogTable, it may contains all the computed columns defined in DDL. But not all of them are retained in source, some of them may be applied to the following LogicalProject. Could we only put the information we need in TabeSourceTable and TableSinkTable? e.g. List<String> partitionKeys.

I would say let's use the information from CatalogTable more carefully then... I can imagine partition keys are just a starter, we might need more and more information from CatalogTable in the future. CatalogTable is some kind of meta information about the table.

Hi @KurtYoung , After temporary table support, every table should have CatalogTable. I think we can keep not Option.
Hi @wuchong , I agree with kurt, Actually, source may also need to know compute columns. It needs to know which fields it does not read. (except compute columns expressions)

source may also need to know compute columns. It needs to know which fields it does not read. (except compute columns expressions)

Actually, we will retain some expression in source (for rowtime generation), then we have to have another field in TableSourceTable to keep this information. say generatedExpressions, this might be confused with the expressions in CatalogTable?

Hi Jark, I don't quite understand the expression in this source, but I think we need a good name to clarify it. It's may not just the TableSourceTable that saves them together? I think they are all attributes of table.

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

KurtYoung · 2019-10-24T11:23:38Z

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

        getTableSink(identifier).map(sink => {
-          TableSinkUtils.validateSink(catalogSink, identifier, sink)
+          val partKeys =
+            catalogManager.getTable(identifier).get().asInstanceOf[CatalogTable].getPartitionKeys


Should we check the catalog type before cast to CatalogTable? For example, it would be a CatalogView when get table from catalog manager.

getTableSink(identifier) already has this check, I will change the return type of getTableSink and get CatalogTable directly.

...table-planner-blink/src/main/scala/org/apache/flink/table/planner/sinks/TableSinkUtils.scala

...org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.scala

.../main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecSinkRule.scala

...ain/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecSinkRule.scala

KurtYoung · 2019-10-24T11:43:01Z

...ink-table-planner/src/main/scala/org/apache/flink/table/api/internal/BatchTableEnvImpl.scala

        // translate the Table into a DataSet and provide the type that the TableSink expects.
        val result: DataSet[T] = translate(table)(outputType)
        // Give the DataSet to the TableSink to emit it.
-        batchSink.emitDataSet(shuffleByPartitionFieldsIfNeeded(batchSink, result))


why deleting these logic?

first, The former logic is to add shuffle no matter whether it is a static partition or not.
second, after we remove getPartitionFields from PartitionableTableSink, here, we can not get the partitions fields. At present, I don't want to optimize dynamic partition shuffle on legacy planner again.

KurtYoung · 2019-10-24T11:46:01Z

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

-            case partitionableSink: PartitionableTableSink
-              if partitionableSink.getPartitionFieldNames != null
-                && partitionableSink.getPartitionFieldNames.nonEmpty =>
+            case partitionableSink: PartitionableTableSink =>


see comment on BatchExecSink, should we just put the validation here? (For partitioned table, we need to make sure the sink is PartitionableTableSink)

Yes, we can.

I think we have validateSink above, no need to validate here.

...org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.scala

…support in legacy planner

…le to test partition source and sink

JingsongLi · 2019-10-25T02:42:26Z

Rebased resolve conflict and fix comments.

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

KurtYoung

Two commit messages should be adjusted. Not get partition keys from catalog manager, but from catalog table.

KurtYoung · 2019-10-25T06:42:02Z

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala


-  private def getTableSink(objectIdentifier: ObjectIdentifier): Option[TableSink[_]] = {
-    JavaScalaConversionUtil.toScala(catalogManager.getTable(objectIdentifier)) match {
+  private def getTableSink(identifier: ObjectIdentifier): Option[(CatalogTable, TableSink[_])] = {


identifier -> tableIdentifier

KurtYoung · 2019-10-25T06:43:53Z

...org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.scala

+    val tableSourceTable = table.unwrap(classOf[TableSourceTable[_]])
+
+    if (!tableSourceTable.tableSource.isInstanceOf[PartitionableTableSource]) {
+      throw new TableException(s"Table(${table.getQualifiedName}) with partition keys" +


It the table is partitioned, but we have a non PartitionableTableSource, couldn't we just skip partition prune and read the whole data instead?
Throwing an exception doesn't seem to be right.

Although I don't think it's possible to use it like this, it can.

KurtYoung · 2019-10-25T06:45:35Z

.../main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecSinkRule.scala

+            }
+          }
+        case _ => throw new TableException(
+          s"Table(${sinkNode.sinkName}) with partition keys should be a PartitionableTableSink.")


We need PartitionableTableSink to write data to partitioned table: $tableName

KurtYoung · 2019-10-25T06:47:54Z

.../main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecSinkRule.scala

+      sinkNode.sink match {
+        case partitionSink: PartitionableTableSink =>
+          val partKeys = sinkNode.catalogTable.getPartitionKeys
+          if (!partKeys.isEmpty) {


we can assert part keys are non empty?

isPartitioned already judge it, I should remove this judge.

KurtYoung · 2019-10-25T06:49:05Z

.../main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecSinkRule.scala

+          if (!partKeys.isEmpty) {
+            val partitionIndices =
+              partKeys.map(partitionSink.getTableSchema.getFieldNames.indexOf(_))
+            // validate


don't have to validate again? we can move all validation logic to TableSinkUtils:validate?

partKeys and TableSchema are come from catalog manager, they already validate by catalog manager, I think I can remove it. (Not come from sink or source)

KurtYoung · 2019-10-25T06:49:34Z

.../main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecSinkRule.scala

-        requiredTraitSet = requiredTraitSet.plus(
-          FlinkRelDistribution.hash(partitionIndices
-            .map(Integer.valueOf), requireStrict = false))
+            requiredTraitSet = requiredTraitSet.plus(


move this into if (partitionSink.configurePartitionGrouping(true)) {

Now, hash shuffle will add anyway. configurePartitionGrouping only control sort.
In streaming mode, will add hash shuffle, but not sort for configurePartitionGrouping.

KurtYoung · 2019-10-25T06:50:02Z

...ain/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecSinkRule.scala

-              s"${partitionFields.get(idx)} must be in the schema.")
-          }
-        }
+    if (sinkNode.catalogTable != null && sinkNode.catalogTable.isPartitioned) {


same comments as BatchExecSinkRule

KurtYoung · 2019-10-25T06:51:09Z

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

-        val sinkProperties = catalogTable.toProperties
-        Option(TableFactoryService.find(classOf[TableSinkFactory[_]], sinkProperties)
+        val sinkProperties = table.toProperties
+        Option(table, TableFactoryService.find(classOf[TableSinkFactory[_]], sinkProperties)


Could you explain what's the difference between TableFactoryUtil.createTableSinkForCatalogTable and TableFactoryService.find(classOf[TableSinkFactory[_]], sinkProperties)?

See Catalog.getTableFactory.
Option 1 use Catalog.getTableFactory to get TableFactory and create sink.
Option 2 use TableFactoryService to create TableFactory and create sink.
The reason why we need option 1 is that hive table factory can not find by TableFactoryService, but I think this can be improved in future.

…ble for PartitionableTableSink

JingsongLi · 2019-10-25T08:25:43Z

Two commit messages should be adjusted. Not get partition keys from catalog manager, but from catalog table.

Thanks for your review, updated.

KurtYoung

I have some final comments, the old logic are really a mess.

KurtYoung · 2019-10-25T08:34:43Z

...org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.scala

+
+    val partitionFieldNames = tableSourceTable.catalogTable.getPartitionKeys.toSeq.toArray[String]
+
+    if (!partitionFieldNames.isEmpty) {


I think we don't need this, and it's a really big if statement.

Missed this one.

KurtYoung · 2019-10-25T08:39:44Z

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

-              if partitionableSink.getPartitionFieldNames != null
-                && partitionableSink.getPartitionFieldNames.nonEmpty =>
+            case partitionableSink: PartitionableTableSink =>
              partitionableSink.setStaticPartition(catalogSink.getStaticPartitions)


I'm wondering why we set the static partition information here but not during StreamExecSinkRule or BatchExecSinkRule.

Maybe just for the test in PartitionableSinkITCase.testInsertWithStaticPartitions, it get staticPartitions from the origin Sink, if in ExecSinkRule, it will be another copied sink.
I think we can modify this in #9796

…ble for PartitionableTableSource

…artitionableTableSource and PartitionableTableSink

KurtYoung

LGTM now, +1

JingsongLi mentioned this pull request Oct 16, 2019

[FLINK-14253][table-planner-blink] Add hash distribution and sort grouping only when dynamic partition insert #9796

Merged

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

docete reviewed Oct 18, 2019

View reviewed changes

JingsongLi force-pushed the removeGetPartitionSourceSink branch from f20e67a to 3c1480f Compare October 24, 2019 03:35

JingsongLi force-pushed the removeGetPartitionSourceSink branch from 3c1480f to e3ba6b1 Compare October 24, 2019 07:00

wuchong reviewed Oct 24, 2019

View reviewed changes

...link-table-common/src/main/java/org/apache/flink/table/sources/PartitionableTableSource.java Show resolved Hide resolved

JingsongLi force-pushed the removeGetPartitionSourceSink branch from e3ba6b1 to e5c0586 Compare October 24, 2019 10:22

KurtYoung requested changes Oct 24, 2019

View reviewed changes

JingsongLi added 2 commits October 25, 2019 10:40

[FLINK-14381][table-planner-legacy] Remove dynamic partition shuffle …

565f782

…support in legacy planner

[FLINK-14381][table-planner-blink] User table factory and catalog tab…

6122c86

…le to test partition source and sink

JingsongLi force-pushed the removeGetPartitionSourceSink branch from e5c0586 to 4c59ed7 Compare October 25, 2019 02:42

KurtYoung reviewed Oct 25, 2019

View reviewed changes

...ble-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala Outdated Show resolved Hide resolved

KurtYoung reviewed Oct 25, 2019

View reviewed changes

[FLINK-14381][table-planner-blink] Get partition keys from catalog ta…

92a1014

…ble for PartitionableTableSink

JingsongLi force-pushed the removeGetPartitionSourceSink branch from 4c59ed7 to ca20ef1 Compare October 25, 2019 08:24

JingsongLi closed this Oct 25, 2019

JingsongLi reopened this Oct 25, 2019

KurtYoung reviewed Oct 25, 2019

View reviewed changes

JingsongLi added 2 commits October 25, 2019 17:16

[FLINK-14381][table-planner-blink] Get partition keys from catalog ta…

13fb322

…ble for PartitionableTableSource

[FLINK-14381][table-planner-blink] Remove getPartitionFieldNames in P…

6a82d9f

…artitionableTableSource and PartitionableTableSink

JingsongLi force-pushed the removeGetPartitionSourceSink branch from ca20ef1 to 6a82d9f Compare October 25, 2019 09:17

KurtYoung approved these changes Oct 25, 2019

View reviewed changes

KurtYoung closed this in 335c4f7 Oct 25, 2019

JingsongLi deleted the removeGetPartitionSourceSink branch October 29, 2019 07:33


		val partitionFieldNames = tableSourceTable.catalogTable.getPartitionKeys.toSeq.toArray[String]

		if (!partitionFieldNames.isEmpty) {

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

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

Uh oh!

Conversation

JingsongLi commented Oct 16, 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

Uh oh!

flinkbot commented Oct 16, 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 16, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

docete left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

docete commented Oct 24, 2019

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

KurtYoung commented Oct 24, 2019

Uh oh!

KurtYoung commented Oct 24, 2019

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

KurtYoung commented Oct 24, 2019

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

KurtYoung commented Oct 24, 2019

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

KurtYoung commented Oct 24, 2019

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

KurtYoung commented Oct 24, 2019

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

KurtYoung commented Oct 24, 2019

Uh oh!

JingsongLi commented Oct 24, 2019

Uh oh!

KurtYoung 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

flinkbot commented Oct 16, 2019 •

edited

Loading

flinkbot commented Oct 16, 2019 •

edited

Loading

wuchong Oct 25, 2019 •

edited

Loading