[FLINK-5188] Create analog of RowCsvInputFormat in java and adjust all the imports of Row and RowTypeInfo #3003

tonycox · 2016-12-14T07:23:31Z

Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the How To Contribute guide.
In addition to going through the list, please provide a meaningful description of your changes.

General
- The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text")
- 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

…rator in core

…w and RowTypeInfo imports

fhueske

Thanks for the doing this refactoring @tonycox!
I had a few comments but can fix them myself before merging.

Thanks, Fabian

fhueske · 2016-12-14T09:38:39Z

flink-java/src/main/java/org/apache/flink/api/java/io/RowCsvInputFormat.java

+import org.apache.flink.types.Row;
+import org.apache.flink.types.parser.FieldParser;
+
+@Internal


I we move it here, we can make this@PublicEvolving.

fhueske · 2016-12-14T09:59:56Z

...-libraries/flink-table/src/main/scala/org/apache/flink/api/table/BatchTableEnvironment.scala

@@ -168,8 +168,10 @@ abstract class BatchTableEnvironment(
  private[flink] def explain(table: Table, extended: Boolean): String = {
    val ast = table.getRelNode
    val optimizedPlan = optimize(ast)
-    val dataSet = translate[Row](optimizedPlan)(TypeExtractor.createTypeInfo(classOf[Row]))
-    dataSet.output(new DiscardingOutputFormat[Row])
+    val dataSet = translate[org.apache.flink.types.Row](optimizedPlan) (


Add an import for Row instead of using the full class name.

OK, I see why you did it this way. I'll move this PR behind #3004.

fhueske · 2016-12-14T10:00:11Z

...libraries/flink-table/src/main/scala/org/apache/flink/api/table/StreamTableEnvironment.scala

@@ -327,7 +327,9 @@ abstract class StreamTableEnvironment(
  def explain(table: Table): String = {
    val ast = table.getRelNode
    val optimizedPlan = optimize(ast)
-    val dataStream = translate[Row](optimizedPlan)(TypeExtractor.createTypeInfo(classOf[Row]))
+    val dataStream = translate[org.apache.flink.types.Row](optimizedPlan)(


Add import for Row

fhueske · 2016-12-14T10:03:29Z

...ibraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/RowComparator.scala

@@ -28,6 +28,7 @@ import org.apache.flink.api.table.typeutils.RowComparator.{createAuxiliaryFields
 import org.apache.flink.core.memory.{DataInputView, DataOutputView, MemorySegment}
 import org.apache.flink.types.KeyFieldOutOfBoundsException

+


Unnecessary change

fhueske · 2016-12-14T10:03:52Z

...ibraries/flink-table/src/main/scala/org/apache/flink/api/table/typeutils/TypeConverter.scala


 import scala.collection.JavaConversions._

 object TypeConverter {

-  val DEFAULT_ROW_TYPE = new RowTypeInfo(Seq()).asInstanceOf[TypeInformation[Any]]
+  val DEFAULT_ROW_TYPE = new org.apache.flink.api.java.typeutils.RowTypeInfo(


Import RowTypeInfo

fhueske · 2016-12-14T10:10:16Z

...es/flink-table/src/test/scala/org/apache/flink/api/table/expressions/SqlExpressionTest.scala

@@ -166,5 +166,5 @@ class SqlExpressionTest extends ExpressionTestBase {
  override def testData: Any = new Row(0)

  override def typeInfo: TypeInformation[Any] =
-    new RowTypeInfo(Seq()).asInstanceOf[TypeInformation[Any]]
+    new RowTypeInfo(Seq.empty[TypeInformation[Any]]: _*).asInstanceOf[TypeInformation[Any]]


Can this be changed to new RowTypeInfo()?

fhueske · 2016-12-14T10:10:36Z

...k-table/src/test/scala/org/apache/flink/api/table/expressions/utils/ExpressionTestBase.scala

@@ -117,14 +117,14 @@ abstract class ExpressionTestBase {
    // compile and evaluate
    val clazz = new TestCompiler[MapFunction[Any, String]]().compile(genFunc)
    val mapper = clazz.newInstance()
-    val result = mapper.map(testData).asInstanceOf[Row]
+    val result = mapper.map(testData).asInstanceOf[org.apache.flink.types.Row]


Import Row

fhueske · 2016-12-14T10:11:35Z

...ries/flink-table/src/test/scala/org/apache/flink/api/table/typeutils/RowComparatorTest.scala

 import org.apache.flink.api.table.typeutils.RowComparatorTest.MyPojo
+import org.apache.flink.api.java.typeutils.{RowTypeInfo => RowTypeInfoNew}


Why do we need to rename here?

fhueske · 2016-12-14T10:11:47Z

...le/src/test/scala/org/apache/flink/api/table/typeutils/RowComparatorWithManyFieldsTest.scala

@@ -21,7 +21,8 @@ package org.apache.flink.api.table.typeutils
 import org.apache.flink.api.common.ExecutionConfig
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
 import org.apache.flink.api.common.typeutils.{ComparatorTestBase, TypeComparator, TypeSerializer}
-import org.apache.flink.api.table.Row
+import org.apache.flink.types.Row
+import org.apache.flink.api.java.typeutils.{RowTypeInfo => RowTypeInfoNew}


Why renaming here?

fhueske · 2016-12-14T10:12:18Z

...ries/flink-table/src/test/scala/org/apache/flink/api/table/typeutils/RowSerializerTest.scala

@@ -50,8 +50,8 @@ class RowSerializerTest {

  @Test
  def testRowSerializer(): Unit = {
-    val rowInfo: TypeInformation[Row] = new RowTypeInfo(
-      Seq(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
+    val rowInfo = new org.apache.flink.api.java.typeutils.RowTypeInfo(


Import RowTypeInfo

tonycox · 2016-12-14T12:00:16Z

Hi @fhueske okay, as you wish

…lls for new Row type. - Port RowCsvInputFormat to Java and move it to flink-core. This closes apache#3003.

fhueske · 2016-12-15T10:09:30Z

Merging

…lls to new Row type. - Port RowCsvInputFormat to Java and move it to flink-core. This closes apache#3003.

tonycox changed the title ~~[Flink-5188] Create analog of RowCsvInputFormat in java and adjust all the imports of Row and RowTypeInfo~~ [FLINK-5188] Create analog of RowCsvInputFormat in java and adjust all the imports of Row and RowTypeInfo Dec 14, 2016

tonycox force-pushed the FLINK-5188 branch 2 times, most recently from 3317c2b to 0edfd48 Compare December 14, 2016 07:53

wuchong and others added 6 commits December 14, 2016 13:44

[FLINK-5187] [core] Create analog of Row and RowTypeInfo and RowCompa…

53044f1

…rator in core

address review comments and add RowTypeInfoTest

d0ee79b

fix tabs

0b46a75

address review comments

3f6bff3

remove unused ArrayList import

a88a8f4

[FLINK-5188] Create analog of RowCsvInputFormat in java and adjust Ro…

b94b5a7

…w and RowTypeInfo imports

tonycox force-pushed the FLINK-5188 branch from 0edfd48 to b94b5a7 Compare December 14, 2016 09:44

fhueske reviewed Dec 14, 2016

View reviewed changes

fhueske pushed a commit to fhueske/flink that referenced this pull request Dec 14, 2016

[FLINK-5188] [table] [connectors] [core] Adjust imports and method ca…

7b90001

…lls for new Row type. - Port RowCsvInputFormat to Java and move it to flink-core. This closes apache#3003.

asfgit closed this in 4d27f8f Dec 15, 2016

joseprupi pushed a commit to joseprupi/flink that referenced this pull request Feb 12, 2017

[FLINK-5188] [table] [connectors] [core] Adjust imports and method ca…

85565dc

…lls to new Row type. - Port RowCsvInputFormat to Java and move it to flink-core. This closes apache#3003.

hequn8128 pushed a commit to hequn8128/flink that referenced this pull request Jun 22, 2017

[FLINK-5188] [table] [connectors] [core] Adjust imports and method ca…

9e8b97b

…lls to new Row type. - Port RowCsvInputFormat to Java and move it to flink-core. This closes apache#3003.

rmetzger added the component=API/TableSQL label Mar 14, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-5188] Create analog of RowCsvInputFormat in java and adjust all the imports of Row and RowTypeInfo #3003

[FLINK-5188] Create analog of RowCsvInputFormat in java and adjust all the imports of Row and RowTypeInfo #3003

tonycox commented Dec 14, 2016 •

edited

fhueske left a comment

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

fhueske Dec 14, 2016

tonycox commented Dec 14, 2016

fhueske commented Dec 15, 2016

		@@ -28,6 +28,7 @@ import org.apache.flink.api.table.typeutils.RowComparator.{createAuxiliaryFields
		import org.apache.flink.core.memory.{DataInputView, DataOutputView, MemorySegment}
		import org.apache.flink.types.KeyFieldOutOfBoundsException

		import org.apache.flink.api.table.typeutils.RowComparatorTest.MyPojo
		import org.apache.flink.api.java.typeutils.{RowTypeInfo => RowTypeInfoNew}

[FLINK-5188] Create analog of RowCsvInputFormat in java and adjust all the imports of Row and RowTypeInfo #3003

[FLINK-5188] Create analog of RowCsvInputFormat in java and adjust all the imports of Row and RowTypeInfo #3003

Conversation

tonycox commented Dec 14, 2016 • edited

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tonycox commented Dec 14, 2016

fhueske commented Dec 15, 2016

tonycox commented Dec 14, 2016 •

edited