Checking the rowtime and proctime are not allowed empty string in StreamTableEnvironment#validateAndExtractTimeAttributes when called registerDataStreamInternal method, such as below code (added on line 485):

// rowtime and proctime are not allowed empty string
if (rowtime.isDefined && rowtime.get._2.trim == "")

if (proctime.isDefined && rowtime.get._2.trim == "")

Is it right?

Yes, I would also check for null and change the message to "The name of the XXXX attribute must not be empty.".

I think we should allow null to indicate that no time attribute should be added.
Otherwise we need TableSource implementations for all combinations of none, event-time, processing-time, and both.

Timo & Fabian,

I have 2 problems to confirm.

(1) I think this problem only appeared when called StreamTableSourceTable#getRowType method, but not StreamTableEnvironment#validateAndExtractTimeAttributes, so we should check in the method of getRowType, line 42 and 50, is it right?

(2) When check timeSource.getRowtimeAttribute or timeSource.getProctimeAttribute is null or not, (line 42 and 50), we should add codes to check whether 'XXXXAttribute' is empty string, if yes, we can treat it as None. Such as following codes:

//case timeSource: DefinedRowtimeAttribute if timeSource.getRowtimeAttribute != null =>
case timeSource: DefinedRowtimeAttribute if timeSource.getRowtimeAttribute != null && !timeSource.getRowtimeAttribute.trim.equals("") =>
...
//case timeSource: DefinedProctimeAttribute if timeSource.getProctimeAttribute != null =>
case timeSource: DefinedProctimeAttribute if timeSource.getProctimeAttribute != null && !timeSource.getProctimeAttribute.trim.equals("") =>

is it right?

Thanks.

Hi Zhe Li,

yes, we need to fix this issue in StreamTableSourceTable and not in StreamTableEnvironment.
However, I don't think we should treat an empty string the same way as null. Instead I would throw a TableException if the string is empty.

Fabian，

I understand what you mean.

So I modified the codes in StreamTableSourceTable#getRowType to:

  val rowtime = tableSource match {
      case emptyStringTimeSource : DefinedRowtimeAttribute if emptyStringTimeSource.getRowtimeAttribute.trim.equals("")  =>
        throw TableException("The name of the rowtime attribute must not be empty.")

      case timeSource: DefinedRowtimeAttribute if timeSource.getRowtimeAttribute != null  =>
        val rowtimeAttribute = timeSource.getRowtimeAttribute
        Some((fieldCnt, rowtimeAttribute))
      case _ =>
        None
    }

    val proctime = tableSource match {
      case emptyStringTimeSource : DefinedProctimeAttribute if emptyStringTimeSource.getProctimeAttribute.trim.equals("")  =>
        throw TableException("The name of the proctime attribute must not be empty.")

      case timeSource: DefinedProctimeAttribute if timeSource.getProctimeAttribute != null  =>
        val proctimeAttribute = timeSource.getProctimeAttribute
        Some((fieldCnt + (if (rowtime.isDefined) 1 else 0), proctimeAttribute))
      case _ =>
        None
    }

If there is nothing problem, could I fix the issue?

Below is my Jira profile:

Username: lmalds
Full Name: Zhe Li

please assign me the update privilege.

Thanks.

The fix looks good! Please open a pull request with that

I also gave you contributor permissions for the Flink JIRA.
You can now also assign other issues to yourself.

Thank you, Fabian

It's my pleasure

I have opened a pull request just now.

Thank you.

Fixed for 1.3.1 with c7f6d0245cd0d47a99d9badcd43f1ebb8758125e
Fixed for 1.4.0 with 850e4d913bf33d90409a078dab2fbc26bfa976ce