[FLINK-14322][table-api] Add watermark information in TableSchema #9994

wuchong · 2019-10-25T09:52:34Z

What is the purpose of the change

As discussed in FLIP-66, the watermark information should be part of TableSchema, and expose to connectors vis CatalogTable#getTableSchema.

We may need to introduce a WatermarkSpec class to describe watermark information.

Brief change log

Introduce WatermarkSpec class to describe watermark information.
TableSchema#getWatermarkSpecs() returns List<WatermarkSpec>, because we may support multiple watermarks in the future. This is for compatibility purpose. Currently, there is at most one watermark spec in the list.
Introduce SqlExpression which is an internal ResolvedExpression and wraps a SQL expression string. As discussed in FLIP-80 ML, we are planning to use SQL format as the string representation of Expression. However, the path of Expression -> RexNode -> SqlNode -> String still is not fully supported, and there are manay problems that need to be fixed in this path. So SqlExpression is a temporary solution to quickly support FLIP-66 as all the watermark expressions come from SQL DDL.
The properties representation of watermark information:

"schema.watermark.0.rowtime" -> "f1.q2"
"schema.watermark.0.strategy.expression" -> "`f1`.`q2` - INTERVAL '5' SECOND"
"schema.watermark.0.strategy.datatype" -> "TIMESTAMP(3)"

Verifying this change

This change added tests:

Adds TableSchemaTest in flink-table-common to verify watermark relative logic in TableSchema.
Remove TableSchemaTest in flink-table-planner because of class name conflict, and move the tests from it to TableEnvironmentTest and TableSchemaTest in flink-table-common.
Adds testTableSchema in DescriptorPropertiesTest to verify the generated properties key-value string, and the derived TableSchema from properteis.

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-10-25T09:56:53Z

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 1221469 (Wed Dec 04 15:05:40 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

wuchong · 2019-10-25T10:23:21Z

cc @danny0405

flinkbot · 2019-10-25T10:25:57Z

CI report:

05bdcf8 : FAILURE Build
3d71177 : FAILURE Build
6201ece : FAILURE Build
7d95594 : SUCCESS Build
5097f48 : FAILURE Build
9bd15ab : FAILURE Build
d244b4f : FAILURE Build
1221469 : SUCCESS Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

wuchong · 2019-10-30T07:42:46Z

Hi @KurtYoung , do you have time to have a look at this?

KurtYoung · 2019-10-30T07:27:04Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java


 if (fieldNames.length != fieldDataTypes.length) {
- throw new TableException(
+ throw new ValidationException(


we can move such sanity check to Builder?

TableSchema still have a public constructor, even though it is deprecated.

+1 to move all the checks to builder, these sanity check should not happen in the constructor.

As I said above, TableSchema still have a public constructor.

KurtYoung · 2019-10-30T07:28:53Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

-
- // collect indices
- fieldNameToIndex.put(fieldName, i);
+ DataType fieldType = Preconditions.checkNotNull(fieldDataTypes[i]);


ditto, since we have a builder for TableSchema, I think we can move all these basic check into Builder

The same reasone as above. I think we can do this once the public constructor is removed.

KurtYoung · 2019-10-30T07:31:01Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

- "List of all fields: " + Arrays.toString(fieldNames));
+ }
+
+ private void validateAndCreateNameTypeMapping(String fieldName, DataType fieldType, List<String> parentNames) {


List<String> parentNames can be a simple string indicating the fieldNamePrefix.

KurtYoung · 2019-10-30T07:31:57Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

- } else {
- uniqueNames.add(fieldName);
+ // validate watermark and rowtime attribute
+ Preconditions.checkState(


You already make sure about this in Builder

Yes. I think we can remove this one.

KurtYoung · 2019-10-30T07:40:54Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

+ LogicalType watermarkOutputType = watermark.getWatermarkStrategy()
+ .getOutputDataType().getLogicalType();
+ if (watermarkOutputType.getTypeRoot() != TIMESTAMP_WITHOUT_TIME_ZONE &&
+ watermarkOutputType.getTypeRoot() != BIGINT) {


why relaxing BIGINT?

I will remove this. We can introduce this when we can support BIGINT as the watermark generation return type end-to-end.

KurtYoung · 2019-10-30T07:41:29Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

@@ -224,13 +254,31 @@ public DataType toRowDataType() {
 return (TypeInformation<Row>) fromDataTypeToLegacyInfo(toRowDataType());
 }

+ /**
+ * Returns an optional of the watermark specification which contains rowtime attribute


an optional -> a list

KurtYoung · 2019-10-30T07:45:48Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

+ * @param rowtimeAttribute the field name as a rowtime attribute, can be a nested field using dot separator.
+ * @param watermarkStrategy the watermark strategy expression
+ */
+ public Builder watermark(String rowtimeAttribute, ResolvedExpression watermarkStrategy) {


If user using this Builder to manually specify watermark, how can he get ResolvedExpression. AFAIK if he uses scala dsl, it can only generate UnresolvedCallExpression or UnresolvedReferenceExpression

Yes. You are right. There are 2 reasons here:

we can't get ExpressionResolver in builder to resolve expression.

TableSchema.Builder is going to be internally used only. As we are planning to remove {{TableSource#getTableSchema}}, users should don't need to create TableSchema anymore.

I added an @Internal annotation on this method.

So besides from DDL way, how to user specify watermark programmatically?

Using the Schema descriptor API.

If we agree that Expression would make the interface more uniform(can be used by table api), can we use just Expression as the argument specification ? Then we can eliminate use to use the SqlExpression for current implementation.

The problem is there are two major types of Expression now, one is unresolved and another one is resolved. We will only have interface (like Java DSL style) for unresolved expression. But the issue is we need some other information which only accessible from resolved one, like data types.

As we discussed in FLIP-80, we will use the serialize (resolved) expression to string, this happens when serialize CatalogTable to properties. And deserialize string to unresolved expression, this happens when create CatalogTable from properties in Catalog. However, we missed that converting unresovled expression to resolved expression requires ExpressionResolver which is not held by Catalog. If the TableSchema builder requires ResolvedExpression, it's impossible to create a TableSchema or CatalogTable for Catalog.

Let me sum up, I agree with @KurtYoung that using ResolveExpression here may can't work in the future, so I would like change it to :

public Builder watermark(String rowtimeAttribute, String watermarkExpression, DataType watermarkExprOutputType) {

The ResolvedExpression getWatermarkStrategy() in WatermarkSpec will be removed, and will expose String getWatermarkExpressionString() and DataType getWatermarkExprOutputType().

What do you think?

I think using plain strings in Java classes should be avoided in general. This is not very object oriented. I understand that constructing ResolvedExpression might be difficult at certain locations but we could also use just Expression. Actually, a SQL string is also just an unresolved expression until it has been parsed and validated.

I'm fine to use Expression here and introduce an unresolved SqlExpression to wrap it.
So does the following signature looks good to you?

public Builder watermark(String rowtimeAttribute, Expression watermarkExpression, DataType watermarkExprOutputType) {

KurtYoung · 2019-10-30T07:47:00Z

BTW, there are still some testing failures.

wuchong · 2019-10-30T08:16:03Z

@KurtYoung Thanks for reviewing , I updated the pull request.

danny0405 · 2019-11-01T01:44:46Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

@@ -54,44 +58,64 @@

 private final DataType[] fieldDataTypes;

- private final Map<String, Integer> fieldNameToIndex;
+ /** Mapping from qualified field name to (nested) field type. */
+ private final Map<String, DataType> typesByName;



fieldNameToType ?

danny0405 · 2019-11-01T01:45:39Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java


 if (fieldNames.length != fieldDataTypes.length) {
- throw new TableException(
+ throw new ValidationException(


+1 to move all the checks to builder, these sanity check should not happen in the constructor.

danny0405 · 2019-11-01T01:49:38Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

+ String fullFieldName = prefixQualifiedName.isEmpty() ? fieldName : prefixQualifiedName + "." + fieldName;
+ DataType oldValue = typesByName.put(fullFieldName, fieldType);
+ if (oldValue != null) {
+ throw new ValidationException("Field names must be unique. Duplicate field: '" + fullFieldName + "'");


oldValue -> oldType

danny0405 · 2019-11-01T01:51:04Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

+ }
+
+ private void validateAndCreateNameTypeMapping(String fieldName, DataType fieldType, String prefixQualifiedName) {
+ String fullFieldName = prefixQualifiedName.isEmpty() ? fieldName : prefixQualifiedName + "." + fieldName;


Can we define another method probably named "generatesFieldName" to wrap the prefix name logic ? It's hard to infer that a "prefixQualifiedName" means.

BTW, i think we can just name this method as "createNameToTypeMapping"

I think the logic of full name generation is simple (just one line), there is no need to extract a method for it. I can add some javadocs to describe the meaning of prefixQualifiedName.

This method is not just creating a mapping, but also does some validation. I think having a "validate" in the method name will be easier to understand.

danny0405 · 2019-11-01T01:55:15Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

@@ -157,11 +181,11 @@ public TableSchema copy() {
 /**
 * Returns the specified data type for the given field name.
 *
- * @param fieldName the name of the field
+ * @param fieldName the name of the field, the field name can be a qualified name, e.g. "f0.q1"
 */
 public Optional<DataType> getFieldDataType(String fieldName) {


"f0.q1" is a little confusing, maybe "record_name.field_name" ?

But what's the difference between record_name and field name?

I changed it to field1.innerField2

danny0405 · 2019-11-01T01:59:41Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

+ * @param rowtimeAttribute the field name as a rowtime attribute, can be a nested field using dot separator.
+ * @param watermarkStrategy the watermark strategy expression
+ */
+ public Builder watermark(String rowtimeAttribute, ResolvedExpression watermarkStrategy) {


If we agree that Expression would make the interface more uniform(can be used by table api), can we use just Expression as the argument specification ? Then we can eliminate use to use the SqlExpression for current implementation.

danny0405 · 2019-11-01T02:01:03Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/WatermarkSpec.java

+ * The referenced attribute must be present in the {@link TableSchema} and of
+ * type {@link DataTypes#TIMESTAMP(int)}.
+ */
+ public String getRowtimeAttribute() {


DataTypes#TIMESTAMP(int) is just a tool method and not a "type".

danny0405 · 2019-11-01T02:03:52Z

...table/flink-table-common/src/main/java/org/apache/flink/table/expressions/SqlExpression.java

+
+/**
+ * Expression that wraps a SQL expression string, e.g. {@code date_format(`f0`, 'yyyy-MM-dd')}.
+ * The SQL expression string can be used as serialized string and can be used for storing in Catalog.


"e.g. {@code date_format(f0, 'yyyy-MM-dd')}" confused me a lot ? Is it a format ?

It is a javadoc DSL like `` in markdown.

danny0405 · 2019-11-01T02:19:01Z

...ble/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala

+ }
+
+ def createRow(innerFieldName: String): TypeInformation[_] = {
+ Types.ROW(


If these 2 methods are only used by this test case, maybe we can eliminate them.

These 2 methods are inner methods and used by multiple times. If we remove them, there will be some code duplication here.

wuchong · 2019-11-01T08:18:23Z

Hi @danny0405 @KurtYoung , I updated the PR.

KurtYoung

LGTM, i left two minor comments.

KurtYoung · 2019-11-01T08:51:28Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/WatermarkSpec.java

+
+ public WatermarkSpec(String rowtimeAttribute, String watermarkExpressionString, DataType watermarkExprOutputType) {
+ this.rowtimeAttribute = checkNotNull(rowtimeAttribute);
+ this.watermarkExpressionString = watermarkExpressionString;


checkNotNull？

KurtYoung · 2019-11-01T08:53:15Z

...table/flink-table-common/src/main/java/org/apache/flink/table/expressions/SqlExpression.java

+ * from {@link ResolvedExpression} to SQL string.
+ */
+@Internal
+public class SqlExpression implements ResolvedExpression {


unnecessary class in this PR?

Good catch! I will introduce it in the following PR.

danny0405 · 2019-11-01T09:08:39Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

 }

 /**
 * Returns a deep copy of the table schema.
 */
 public TableSchema copy() {
- return new TableSchema(fieldNames.clone(), fieldDataTypes.clone());
+ return new TableSchema(fieldNames.clone(), fieldDataTypes.clone(), new ArrayList<>(watermarkSpecs));
 }


Why not just clone the watermarkSpecs ?

Because WatermarkSpec is an immutable object. We don't need to clone an immutable object just like String.

danny0405 · 2019-11-01T09:08:57Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

@@ -157,11 +181,11 @@ public TableSchema copy() {
 /**
 * Returns the specified data type for the given field name.
 *
- * @param fieldName the name of the field
+ * @param fieldName the name of the field, the field name can be a qualified name, e.g. "f0.q1"
 */
 public Optional<DataType> getFieldDataType(String fieldName) {


danny0405 · 2019-11-01T09:11:49Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

+ */
+ private void validateAndCreateNameToTypeMapping(String fieldName, DataType fieldType, String parentFieldName) {
+ String fullFieldName = parentFieldName.isEmpty() ? fieldName : parentFieldName + "." + fieldName;
+ DataType oldType = fieldNameToType.put(fullFieldName, fieldType);


Can we pass null as parentFieldName ?

We shouldn't pass null. If it can accept nullable String, then the parameter should be annotated @Nullable, otherwise, it is not null.
Besides, it is only internally used and we guaranteed that.

danny0405 · 2019-11-01T09:13:07Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java

+ * not be validated by {@link TableSchema}.
+ */
+ public Builder watermark(String rowtimeAttribute, String watermarkExpressionString, DataType watermarkExprOutputType) {
+ Preconditions.checkNotNull(rowtimeAttribute);


watermarkExpressionString -> watermarkExpr, the type already indicates that it is a String;
watermarkExprOutputType -> watermarkExprType obviously the type of a expression is it's output type.

We don't run out of characters. A better field name describes everything. If we use watermarkExpr, some users may guess it is an Expression but is a String.

danny0405 · 2019-11-01T09:15:54Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/WatermarkSpec.java

+ * Watermark metadata defined in {@link TableSchema}. It mainly contains two parts,
+ * 1) the rowtime attribute.
+ * 2) the watermark generation strategy.
+ */


You can use a ol list to list these doc items.

danny0405 · 2019-11-01T09:16:18Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/WatermarkSpec.java

+ private final String rowtimeAttribute;
+
+ private final String watermarkExpressionString;
+


Can we just comment all these members ?

I can copy the javadoc to the members here, but I think it is not necessary. Because the javadoc of the getter methods explain them in detail. These private members are invisible to users.

danny0405 · 2019-11-01T09:17:38Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/WatermarkSpec.java

+ * The string representation is a qualified SQL expression string (UDFs are expanded).
+ */
+ public String getWatermarkExpressionString() {
+ return watermarkExpressionString;


getWatermarkExpression is okey i think.

A String in the method name explicitly declares that the return value is a string representation instead of Expression. It is weird to return String but the method name says it is a expression.

danny0405

+1, LGTM.

wuchong · 2019-11-01T12:38:38Z

Travis passed: https://travis-ci.com/flink-ci/flink/builds/134547319
Merging...

wuchong requested review from twalthr and KurtYoung October 25, 2019 09:52

rmetzger added the review=description? label Oct 25, 2019

rmetzger added component=TableSQL/Planner component=TableSQL/API labels Oct 25, 2019

KurtYoung reviewed Oct 30, 2019

View reviewed changes

danny0405 reviewed Nov 1, 2019

View reviewed changes

KurtYoung approved these changes Nov 1, 2019

View reviewed changes

danny0405 reviewed Nov 1, 2019

View reviewed changes

danny0405 approved these changes Nov 1, 2019

View reviewed changes

[FLINK-14322][table-api] Add watermark information in TableSchema

1221469

wuchong force-pushed the watermark-schema branch from d244b4f to 1221469 Compare November 1, 2019 11:13

wuchong merged commit 3011bff into apache:master Nov 1, 2019

wuchong deleted the watermark-schema branch November 1, 2019 12:39

		private final String rowtimeAttribute;

		private final String watermarkExpressionString;

[FLINK-14322][table-api] Add watermark information in TableSchema #9994

[FLINK-14322][table-api] Add watermark information in TableSchema #9994

Conversation

wuchong commented Oct 25, 2019 • 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 Oct 25, 2019 • edited Loading

Automated Checks

Review Progress

wuchong commented Oct 25, 2019

flinkbot commented Oct 25, 2019 • edited Loading

CI report:

wuchong commented Oct 30, 2019

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

Choose a reason for hiding this comment

wuchong Nov 1, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KurtYoung commented Oct 30, 2019

wuchong commented Oct 30, 2019

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

wuchong Nov 1, 2019 • 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

Choose a reason for hiding this comment

wuchong commented Nov 1, 2019

KurtYoung 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

wuchong Nov 1, 2019 • edited Loading

Choose a reason for hiding this comment

danny0405 left a comment

Choose a reason for hiding this comment

wuchong commented Nov 1, 2019

wuchong commented Oct 25, 2019 •

edited

Loading

flinkbot commented Oct 25, 2019 •

edited

Loading

flinkbot commented Oct 25, 2019 •

edited

Loading

wuchong Nov 1, 2019 •

edited

Loading

wuchong Nov 1, 2019 •

edited

Loading

wuchong Nov 1, 2019 •

edited

Loading