[ 
https://issues.apache.org/jira/browse/FLINK-7548?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16221809#comment-16221809
 ] 

ASF GitHub Bot commented on FLINK-7548:
---------------------------------------

Github user twalthr commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4894#discussion_r146586028
  
    --- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonScan.scala
 ---
    @@ -18,55 +18,7 @@
     
     package org.apache.flink.table.plan.nodes
     
    -import org.apache.flink.api.common.functions.Function
    -import org.apache.flink.api.common.typeinfo.TypeInformation
    -import org.apache.flink.table.api.TableConfig
    -import org.apache.flink.table.codegen.{FunctionCodeGenerator, 
GeneratedFunction}
    -import org.apache.flink.types.Row
    -
     /**
       * Common class for batch and stream scans.
       */
    -trait CommonScan[T] {
    -
    -  /**
    -    * We check if the input type is exactly the same as the internal row 
type.
    -    * A conversion is necessary if types differ.
    -    */
    -  private[flink] def needsConversion(
    -      externalTypeInfo: TypeInformation[Any],
    -      internalTypeInfo: TypeInformation[T]): Boolean =
    -    externalTypeInfo != internalTypeInfo
    -
    -  private[flink] def generatedConversionFunction[F <: Function](
    -      config: TableConfig,
    -      functionClass: Class[F],
    -      inputType: TypeInformation[Any],
    -      expectedType: TypeInformation[Row],
    -      conversionOperatorName: String,
    -      fieldNames: Seq[String],
    -      inputFieldMapping: Option[Array[Int]] = None)
    -    : GeneratedFunction[F, Row] = {
    -
    -    val generator = new FunctionCodeGenerator(
    -      config,
    -      false,
    -      inputType,
    -      None,
    -      inputFieldMapping)
    -    val conversion = 
generator.generateConverterResultExpression(expectedType, fieldNames)
    -
    -    val body =
    -      s"""
    -         |${conversion.code}
    -         |return ${conversion.resultTerm};
    -         |""".stripMargin
    -
    -    generator.generateFunction(
    -      conversionOperatorName,
    -      functionClass,
    -      body,
    -      expectedType)
    -  }
    -
    -}
    +trait CommonScan[T]
    --- End diff --
    
    Do we still need this trait if it does not contain anything? Let's keep it 
simple and remove it for now.


> Support watermark generation for TableSource
> --------------------------------------------
>
>                 Key: FLINK-7548
>                 URL: https://issues.apache.org/jira/browse/FLINK-7548
>             Project: Flink
>          Issue Type: Improvement
>          Components: Table API & SQL
>            Reporter: Jark Wu
>            Assignee: Fabian Hueske
>            Priority: Blocker
>             Fix For: 1.4.0
>
>
> As discussed in FLINK-7446, currently the TableSource only support to define 
> rowtime field, but not support to extract watermarks from the rowtime field. 
> We can provide a new interface called {{DefinedWatermark}}, which has two 
> methods {{getRowtimeAttribute}} (can only be an existing field) and 
> {{getWatermarkGenerator}}. The {{DefinedRowtimeAttribute}} will be marked 
> deprecated.
> How to support periodic and punctuated watermarks and support some built-in 
> strategies needs further discussion.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Reply via email to