[
https://issues.apache.org/jira/browse/FLINK-6039?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15923886#comment-15923886
]
Zhuoluo Yang commented on FLINK-6039:
-------------------------------------
Hi [~fhueske] Thank you for your comments. I'm not sure whether this would be a
default behavior. Maybe a property, an annotation or a subclass could be a new
design.
> Row of TableFunction should support flexible number of fields
> -------------------------------------------------------------
>
> Key: FLINK-6039
> URL: https://issues.apache.org/jira/browse/FLINK-6039
> Project: Flink
> Issue Type: Improvement
> Reporter: Zhuoluo Yang
> Assignee: Zhuoluo Yang
>
> In actual world, especially while processing logs with TableFunction. The
> formats of the logs in actual world are flexible. Thus, the number of fields
> should not be fixed.
> For examples, we should make the three following types of of TableFunction
> work.
> {code}
> // Test for incomplete row
> class TableFunc4 extends TableFunction[Row] {
> def eval(str: String): Unit = {
> if (str.contains("#")) {
> str.split("#").foreach({ s =>
> val row = new Row(3)
> row.setField(0, s) // And we only set values for one column
> collect(row)
> })
> }
> }
> override def getResultType: TypeInformation[Row] = {
> new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.INT_TYPE_INFO,
> BasicTypeInfo.INT_TYPE_INFO)
> }
> }
> // Test for incomplete row
> class TableFunc5 extends TableFunction[Row] {
> def eval(str: String): Unit = {
> if (str.contains("#")) {
> str.split("#").foreach({ s =>
> val row = new Row(1) // ResultType is three columns, we have only
> one here
> row.setField(0, s)
> collect(row)
> })
> }
> }
> override def getResultType: TypeInformation[Row] = {
> new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.INT_TYPE_INFO,
> BasicTypeInfo.INT_TYPE_INFO)
> }
> }
> // Test for overflow row
> class TableFunc6 extends TableFunction[Row] {
> def eval(str: String): Unit = {
> if (str.contains("#")) {
> str.split("#").foreach({ s =>
> val row = new Row(5) // ResultType is two columns, we have five
> columns here
> row.setField(0, s)
> row.setField(1, s.length)
> row.setField(2, s.length)
> row.setField(3, s.length)
> row.setField(4, s.length)
> collect(row)
> })
> }
> }
> override def getResultType: TypeInformation[Row] = {
> new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.INT_TYPE_INFO)
> }
> }
> {code}
> Actually, the TableFunc4 and TableFunc6 has already worked correctly with
> current version. This issue will make TableFunc5 works.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
