flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhueske <...@git.apache.org>
Subject [GitHub] flink pull request #5240: [FLINK-8240] [table] Create unified interfaces to ...
Date Fri, 26 Jan 2018 17:54:05 GMT
Github user fhueske commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5240#discussion_r164151474
  
    --- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/Schema.scala
---
    @@ -69,17 +72,76 @@ class Schema extends Descriptor {
         */
       def field(fieldName: String, fieldType: String): Schema = {
         if (tableSchema.contains(fieldName)) {
    -      throw new IllegalArgumentException(s"Duplicate field name $fieldName.")
    +      throw new ValidationException(s"Duplicate field name $fieldName.")
    +    }
    +
    +    val fieldProperties = mutable.LinkedHashMap[String, String]()
    +    fieldProperties += (DescriptorUtils.TYPE -> fieldType)
    +
    +    tableSchema += (fieldName -> fieldProperties)
    +
    +    lastField = Some(fieldName)
    +    this
    +  }
    +
    +  /**
    +    * Specifies the origin of the previously defined field. The origin field is defined
by a
    +    * connector or format.
    +    *
    +    * E.g. field("myString", Types.STRING).from("CSV_MY_STRING")
    +    */
    +  def from(originFieldName: String): Schema = {
    +    lastField match {
    +      case None => throw new ValidationException("No field defined previously. Use
field() before.")
    +      case Some(f) =>
    +        tableSchema(f) += (DescriptorUtils.FROM -> originFieldName)
    +        lastField = None
    +    }
    +    this
    +  }
    +
    +  /**
    +    * Specifies the previously defined field as a processing-time attribute.
    +    *
    +    * E.g. field("myString", Types.STRING).proctime()
    +    */
    +  def proctime(): Schema = {
    +    lastField match {
    +      case None => throw new ValidationException("No field defined previously. Use
field() before.")
    +      case Some(f) =>
    +        tableSchema(f) += (DescriptorUtils.PROCTIME -> DescriptorUtils.TRUE)
    +        lastField = None
    +    }
    +    this
    +  }
    +
    +  /**
    +    * Specifies the previously defined field as an event-time attribute.
    +    *
    +    * E.g. field("myString", Types.STRING).rowtime(...)
    --- End diff --
    
    `field("procTime", Types.SQL_TIMESTAMP).proctime()`


---

Mime
View raw message