[FLINK-7446] [table] Change DefinedRowtimeAttribute to work on existing field. #4710

fhueske · 2017-09-23T23:03:06Z

What is the purpose of the change

Changes the contract of the DefinedRowtimeAttribute interface. The rowtime attribute is no longer appended to the schema of the row but marks an existing field in the input that will be handled as event time attribute. The specified field must be of type Long or Timestamp. The watermarks of DataStream must be aligned to the specified field.

Brief change log

rowtime fields are no longer appended but an existing Long or Timestamp field is marked as event time field
Add checks that projections are not pushed to TableSources that implement DefinedRowtimeAttribute or DefinedProctimeAttribute
Added several tests for table sources with rowtime or proctime attributes
Updated the documenation

Verifying this change

Check the added test cases

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

Documentation

Documentation is updated according to the changes of the PR.

…ng field.

haohui · 2017-09-24T07:42:44Z

LGTM overall +1.

One question: since we now cast ROWTIME / PROCTIME directly to LONG, I wonder, do we want to revisit the decision that creates dedicated types for ROWTIME / PROCTIME?

twalthr

Thanks for the PR @fhueske. Looks very good! I had just minor comments. One last thing: Could could you also add a combined TableSource that implements batch and stream table source + defined time attributes?

twalthr · 2017-09-25T13:15:37Z

docs/dev/table/streaming.md

 #### During DataStream-to-Table Conversion

-The event time attribute is defined with the `.rowtime` property during schema definition. 
+The event time attribute is defined with the `.rowtime` property during schema definition. Timestamps and watermarks must have been assigned in the `DataStream` that is converted.


Add a link to dev/event_timestamps_watermarks again?

twalthr · 2017-09-25T13:28:09Z

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala

      case idx =>
-        // regular attribute. Access attribute in input data type.
-        generateInputAccess(input1, input1Term, idx)
+        // get type of result field


add a comment that this is needed for TableSource?

twalthr · 2017-09-25T13:28:32Z

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala

+        // Change output type to rowtime indicator
+        if (FlinkTypeFactory.isRowtimeIndicatorType(outType) &&
+          (inputAccess.resultType == Types.LONG || inputAccess.resultType == Types.SQL_TIMESTAMP)) {
+          // Hard cast possible because LONG, TIMESTAMP, and ROW_TIMEINDICATOR are internally


ROWTIME_INDICATOR

twalthr · 2017-09-25T13:31:36Z

...e/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalTableSourceScan.scala

+    tableSource match {
+      case s: StreamTableSource[_] =>
+        StreamTableSourceTable.deriveRowTypeOfTableSource(s, flinkTypeFactory)
+      case b: BatchTableSource[_] =>


Replace b with _ to remove IDE warning.

twalthr · 2017-09-25T13:32:53Z

...ain/scala/org/apache/flink/table/plan/rules/logical/PushProjectIntoTableSourceScanRule.scala

    val scan: FlinkLogicalTableSourceScan = call.rel(1).asInstanceOf[FlinkLogicalTableSourceScan]
    scan.tableSource match {
+      // projection pushdown is not supported for sources that provide time indicators
+      case r: DefinedRowtimeAttribute if r.getRowtimeAttribute != null => false


Can you create a follow-up issue for this?

Yes will do, but it is not easy to solve IMO.

twalthr · 2017-09-25T13:37:13Z

...s/flink-table/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala

    val original = TableEnvironment.getFieldIndices(tableSource)

-    // append rowtime marker
-    val withRowtime = if (rowtime.isDefined) {


remove unused rowtime

twalthr · 2017-09-25T13:40:26Z

...raries/flink-table/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala

-    * Defines a name of the event-time attribute that represents Flink's
-    * event-time. Null if no rowtime should be available.
+    * Defines a name of the event-time attribute that represents Flink's event-time, i.e., an
+    * attribute that is aligned with the watermarks of the table.


that is aligned with the watermarks of the underlying DataStream?

twalthr · 2017-09-25T13:40:55Z

...raries/flink-table/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala

+    * The method should return null if no rowtime attribute is defined.
+    *
+    * @return The name of the field that represents the event-time field and which is aligned
+    *         with the watermarks of the table. The field must be present in the schema of the


twalthr · 2017-09-25T13:41:10Z

...raries/flink-table/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala

 trait DefinedRowtimeAttribute {

  /**
-    * Defines a name of the event-time attribute that represents Flink's


Maybe also update the docs of the class?

twalthr · 2017-09-25T13:54:24Z

...ies/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/TableSourceTest.scala

+  }
+
+  @Test
+  def testProjectableRowTimeTableSource(): Unit = {


Does filter push down work? With rowtime and proctime?

Yes, that should not be a problem. Projection push-down is not possible because the schema of the table is partially constructed inside of the Table API (e.g., appending the proctime attribute).

twalthr · 2017-09-25T14:06:40Z

@haohui We only cast ROWTIME / PROCTIME directly to LONG during runtime, the special types are needed during pre-flight phase and validation. We could not come up with a better solution that ensures that watermarks stay aligned with the rowtime.

fhueske · 2017-09-26T17:54:12Z

Thanks for the review @twalthr.
I've updated the PR.

@haohui: This PR preserves the current logic that time attributes are exposed as TIMESTAMP. I agree that support for time indicators that expose themselves as Long is desirable. However, this requires quite a few changes as we need to extend several functions (incl. TUMBLE, HOP, etc.) and validation logic in some operators (over windows, joins, etc.). So this is not a lightweight change and should be done as a separate issue, IMO.

wuchong · 2017-09-27T16:25:20Z

Looks good to me.

+1

fhueske · 2017-10-04T21:30:15Z

Thanks @wuchong
I will merge this PR tomorrow.

[FLINK-7446] [table] Change DefinedRowtimeAttribute to work on existi…

28f11bc

…ng field.

twalthr requested changes Sep 25, 2017

View reviewed changes

addressed feedback

796a2be

asfgit closed this in dae21da Oct 5, 2017

fhueske deleted the tableExistingField branch October 5, 2017 07:13

rmetzger added the component=API/TableSQL label Mar 18, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

[FLINK-7446] [table] Change DefinedRowtimeAttribute to work on existing field. #4710

[FLINK-7446] [table] Change DefinedRowtimeAttribute to work on existing field. #4710

Uh oh!

Conversation

fhueske commented Sep 23, 2017

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!

haohui commented Sep 24, 2017

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!

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!

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!

twalthr commented Sep 25, 2017

Uh oh!

fhueske commented Sep 26, 2017

Uh oh!

wuchong commented Sep 27, 2017

Uh oh!

fhueske commented Oct 4, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants