[FLINK-6841][table]Using TableSourceTable for both Stream and Batch

[sunjincheng121]

Remove StreamTableSourceTable, And using TableSourceTable for both Stream and Batch.

• General

  • The pull request references the related JIRA issue ("[FLINK-6841][table]Using TableSourceTable for both Stream and Batch")
  • The pull request addresses only one issue
  • Each commit in the PR has a meaningful commit message (including the JIRA id)

• Documentation

  • Documentation has been added for new functionality
  • Old documentation affected by the pull request has been updated
  • JavaDoc for public methods has been added

• Tests & Build

  • Functionality added by the pull request is covered by tests
  • mvn clean verify has been executed successfully locally or a Travis build has passed

[twalthr]

I'm not sure about this change. A batch table source would be able to create a time attribute if it implements DefinedProctimeAttribute, right?

[sunjincheng121]

Hi @twalthr I think DefinedProctimeAttribute can be implemented by a batch table source. If we do not want a batch table source can implements DefinedProctimeAttribute , we should move DefinedProctimeAttribute to StreamTableSourceTable. Otherwise, I feel that StreamTableSourceTable can be deleted.Just like the changes of this PR. What do you think?

[wuchong]

I think checking the result of fromExternalCatalogTable is a TableSourceTable is meaningless now (it definitely returns a TableSourceTable). I suggest to check whether the TableSource in the TableSourceTable is a StreamTableSource.

assertTrue(tableSource.tableSource.isInstanceOf[StreamTableSource[_]])

[sunjincheng121]

Make sense to me. :)

[wuchong]

@twalthr @sunjincheng121 DefinedProctimeAttribute and DefinedRowtimeAttribute doesn't work in batch table source, currently. BTW, I don't think batch table source needs DefinedRowtimeAttribute. Users can use an existing column as the row time attribute without declare the column as a rowtime, isn't that right ?

For the following example, the ts is a physical column in table T.

SELECT overAgg(b, a) FROM T GROUP BY TUMBLE(ts, INTERVAL '2' HOUR)`

[sunjincheng121]

Hi, @wuchong Thanks for your reviewing. I meant from the perspective of grammar is allow batch table source to implement the DefinedProctimeAttribute and DefinedRowtimeAttribute. We are on the same page about batch table source do not needs DefinedRowtimeAttribute.

I have updated the PR. according your comment. @wuchong

Thanks,
SunJincheng

[fhueske]

Hi @sunjincheng121! I left a comment on the JIRA issue. Short summary: I don't think this PR is a significant improvement. I'd rather keep it as it is unless you have a good argument to convince me.

Thank you, Fabian

[sunjincheng121]

Thanks @fhueske, I noticed your description on the JIRA. issue. I think it's make sense to me. Because at the current time between StreamTableSourceTable and (Batch)TableSourceTable really some different, such as watermarks, time attributes etc. which you have mentioned. So I agree with you, and close both the PR. and JIRA.

Thanks,
SunJincheng