[FLINK-5187] [core] Create analog of Row and RowTypeInfo and RowCompator in core #2968

wuchong · 2016-12-08T14:54:17Z

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

twalthr · 2016-12-08T16:38:22Z

Thanks for working on this @wuchong. I will review this next week.

tonycox

Hi @wuchong, Thanks for the PR.
I left some minor comments

tonycox · 2016-12-08T16:03:55Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java

+	public int getFieldIndex(String fieldName) {
+		for (int i = 0; i < fieldNames.length; i++) {
+			if (fieldNames[i].equals(fieldName)) {
+				return i;


what if we have more than field one with the same name?

The field names of Row is forced to be f0~fn.

tonycox · 2016-12-08T16:14:07Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java

+		this.fieldNames = new String[types.length];
+
+		for (int i = 0; i < types.length; i++) {
+			fieldNames[i] = "f" + i;


Can we somehow get real names of fields?

I think RowTypeInfo is something like TupleTypeInfo, which can't have custom field names.

Actually, we had custom field names before and removed them at some point.
It might make sense to add support for optional names again now that Row is becoming more user-facing. However, I think we can do this as a separate issue.

tonycox · 2016-12-08T16:20:43Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java

+	@Override
+	public void writeWithKeyNormalization(
+		Row record,
+		DataOutputView target) throws IOException {


Can be in one line

tonycox · 2016-12-08T16:21:05Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java

+	public int extractKeys(Object record, Object[] target, int index) {
+		int len = comparators.length;
+		int localIndex = index;
+		for(int i=0;i<len;i++){


Please, add some spaces

tonycox · 2016-12-08T16:24:44Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java

+	/**
+	 * @return creates auxiliary fields for normalized key support
+	 */
+	private static Tuple4<int[], Integer, Integer, Boolean>


I think it's the same as <int[], int, int, boolean>

Unfortunately, the generic type of Tuple can not be primitive types.

tonycox · 2016-12-08T16:31:31Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+	private static final long serialVersionUID = 1L;
+
+	/** Number of field. */
+	private final int arity;


fields.length is similar to arity. Maybe should avoid this field

Good point!

tonycox · 2016-12-08T16:36:15Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java

+	/**
+	 * Intermediate constructor for creating auxiliary fields.
+	 */
+	private RowComparator(


I think this constructor is redundant. int[] normalizedKeyLengths, int numLeadingNormalizableKeys, int normalizableKeyPrefixLen, boolean invertNormKey can be initialized direct form Tuple4<>

This constructor is used in duplicate which I don't want to generate the Tuple4 again.

wuchong · 2016-12-09T03:44:08Z

Hi @tonycox , thanks for your reviewing. I addressed some of your comments.

tonycox · 2016-12-09T07:38:07Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java

-				normalizableKeyPrefixLen == Integer.MAX_VALUE ||
-				normalizableKeyPrefixLen > keyBytes;
+		       normalizableKeyPrefixLen == Integer.MAX_VALUE ||
+		       normalizableKeyPrefixLen > keyBytes;


here must be only tabs instead of spaces

tonycox · 2016-12-09T08:18:55Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java

+	/** Temporary variable for directly passing orders to comparators. */
+	private boolean[] comparatorOrders = null;
+
+	public RowTypeInfo(TypeInformation<?>... types) {


Please add another one constructor with Collection<TypeInformation<?>> types to set for example Seq(Types.INT, Types.LONG, Types.STRING) from scala

Actually, Seq is not a subclass of Collection, provide Collection<TypeInformation<?>> types doesn't fix the problem. I think it's better to adapt the Scala code where has used RowTypeInfo in this form: new RowTypeInfo(Types.INT, Types.LONG, Types.STRING). What do you think?

scala.collection.Seq easily converts to java.util.List with import scala.collection.JavaConversions._ in scala code

And when we set Array[TypeInformation[_]] in constructor from scala it has conflict with TypeInformation<?>..., so I propose cast it .toList .
What do you think?

I would like the following ways of using in Scala, rather than import JavaConversions._ or manually cast it with .toList. In this way, we only need TypeInformation<?>... types constructor.

new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.BIG_DEC_TYPE_INFO) new RowTypeInfo(Seq(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.BIG_DEC_TYPE_INFO): _*) new RowTypeInfo(Array(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.BIG_DEC_TYPE_INFO): _*)

tonycox · 2016-12-09T16:32:26Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+
+	@Override
+	public String toString() {
+		return Arrays.deepToString(fields);


What do you think about replacing all '[' and ']' from string?

In Row tests expected = 1,3,2,1,3,Hello world, but actual = [1, 3, 2, 1, 3, Hello world]

Which "Row test" do you mean ?

For example org.apache.flink.api.scala.stream.table.AggregationsITCase or org.apache.flink.api.scala.batch.table.JoinITCase

You are right. The IT cases use Row as the result type and check the toString results. But IMO, the format 1,3,2,1,3,Hello world without brackets is not a good design, which is hard to distinguish the boundary. And the format [1, 3, 2, 1, 3, Hello world] with square brackets is not good too, because it looks like an array.

I would like to propose the format like Scala Case Class and Flink Java Tuple 's toString, which should be (1,3,2,1,3,Hello world). And we should also adapt the IT cases to the new Row's toString method. What do you think @tonycox @twalthr ?

Agree, the toString method should return the same representation as before.

tonycox · 2016-12-12T08:39:12Z

Hi @wuchong, maybe we should move RowCsvInputFormat also

wuchong · 2016-12-12T08:51:29Z

Hi @tonycox , I agree we should move RowCsvInputFormat also. But I think it would be better to do that in another JIRA like FLINK-5188. Let's keep this PR simple so that it can be in before 1.2 release . What do you think ?

fhueske

Thanks for porting Row and all related classes to Java and moving them to fflink-core.
I made just a few style comments. Looks good otherwise, IMO.

We should merge this rather soon because this change should go into 1.2 and we need a few follow-ups on this.

fhueske · 2016-12-09T17:01:15Z

docs/dev/types_serialization.md

@@ -92,6 +92,8 @@ Internally, Flink makes the following distinctions between types:

  * POJOs: classes that follow a certain bean-like pattern

+  * Row (unlimited length of fields)


Tuples with arbitrary number of fields and support for null fields.

Extend the description of the Java and Scala tuples (max 25 / 22 fields, null fields not supported) to make the difference to Row clear.

fhueske · 2016-12-09T17:23:19Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java

+		this.fieldNames = new String[types.length];
+
+		for (int i = 0; i < types.length; i++) {
+			fieldNames[i] = "f" + i;


Actually, we had custom field names before and removed them at some point.
It might make sense to add support for optional names again now that Row is becoming more user-facing. However, I think we can do this as a separate issue.

fhueske · 2016-12-09T17:25:02Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/RowTypeInfo.java

+		boolean[] orders,
+		int logicalFieldOffset,
+		ExecutionConfig config) {
+		comparatorOrders = orders;


can you add a new line here to separate parameters from the method body?

fhueske · 2016-12-09T17:44:12Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullMaskUtils.java

+		int len,
+		DataInputView source,
+		boolean[] nullMask) throws IOException {
+		int b = 0x00;


fhueske · 2016-12-10T20:12:21Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java

+		TypeSerializer<Object>[] serializers,
+		int[] normalizedKeyLengths,
+		int numLeadingNormalizableKeys, int normalizableKeyPrefixLen, boolean invertNormKey) {
+		this.arity = arity;


fhueske · 2016-12-12T13:33:08Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+	}
+
+	/**
+	 * Get the number of field in the Row.


field -> fields

fhueske · 2016-12-12T13:34:54Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+
+	@Override
+	public String toString() {
+		return Arrays.deepToString(fields);


Agree, the toString method should return the same representation as before.

fhueske · 2016-12-12T13:43:19Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+
+		Row row = (Row) o;
+
+		// Probably incorrect - comparing Object[] arrays with Arrays.equals


What do you mean by this comment?

It's a IDEA code generation comment. We can remove this .

fhueske · 2016-12-12T13:52:42Z

...c/test/java/org/apache/flink/api/common/typeutils/base/RowComparatorWithManyFieldsTests.java

+		createRow("a3", "b3", "c3", "d3", "e3", "f3", "g3", "h3", "i3", "j3")
+	};
+
+	@Before


Can be @BeforeClass

fhueske · 2016-12-12T13:53:05Z

flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/RowComparatorTest.java

+		createRow(1, 1.0, "b", new Tuple3<>(2, true, (short) 3), testPojo3)
+	};
+
+	@Before


Can be @BeforeClass

fhueske · 2016-12-12T14:12:17Z

@tonycox @wuchong Yes, we can move RowCsvInputFormat in FLINK-5188. But that issue should also be included in the 1.2 release. @tonycox, can you start with FLINK-5188 and base it on the changes of this PR? We should try to get this done as soon as possible. Thanks!

wuchong · 2016-12-12T14:51:03Z

Hi @fhueske , thanks for reviewing. I have addressed the comments.

tonycox · 2016-12-12T19:43:03Z

@fhueske Yes. I already started https://github.com/tonycox/flink/tree/FLINK-5188

fhueske

Thanks for the update @wuchong!

Just noticed one more unnecessary import. Otherwise good to merge, IMO. What do you think @twalthr?

fhueske · 2016-12-12T20:54:21Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowComparator.java

+import org.apache.flink.types.Row;
+
+import java.io.IOException;
+import java.util.ArrayList;


ArrayList is unused and needs to be removed.

fhueske · 2016-12-14T11:28:07Z

Just noticed that the row comparator and serializer tests should be moved to org.apache.flink.api.java.typeutils.runtime.

I will do that before merging.

…them to flink-core. This closes apache#2968.

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

merging

…them to flink-core. This closes apache#2968.

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

6b3c98c

…rator in core

tonycox reviewed Dec 8, 2016

View reviewed changes

address review comments and add RowTypeInfoTest

8da708e

tonycox reviewed Dec 9, 2016

View reviewed changes

fix tabs

83bbf76

tonycox reviewed Dec 9, 2016

View reviewed changes

fhueske reviewed Dec 12, 2016

View reviewed changes

address review comments

739af68

fhueske reviewed Dec 12, 2016

View reviewed changes

remove unused ArrayList import

6427023

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

[FLINK-5187] [core] Port Row and related type utils to Java and move …

e130240

…them to flink-core. This closes apache#2968.

asfgit closed this in 86f8a25 Dec 15, 2016

wuchong deleted the row-FLINK-5187 branch December 16, 2016 02:02

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

[FLINK-5187] [core] Port Row and related type utils to Java and move …

3b41d29

…them to flink-core. This closes apache#2968.

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

snyk-bot mentioned this pull request Mar 3, 2022

[Snyk] Upgrade @antv/g2 from 4.1.40 to 4.1.45 MartijnVisser/flink#194

Closed

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

snyk-bot mentioned this pull request Mar 30, 2022

[Snyk] Upgrade @antv/g2 from 4.1.40 to 4.1.47 MartijnVisser/flink#291

Closed

		@@ -92,6 +92,8 @@ Internally, Flink makes the following distinctions between types:

		* POJOs: classes that follow a certain bean-like pattern

		* Row (unlimited length of fields)


		Row row = (Row) o;

		// Probably incorrect - comparing Object[] arrays with Arrays.equals

[FLINK-5187] [core] Create analog of Row and RowTypeInfo and RowCompator in core #2968

[FLINK-5187] [core] Create analog of Row and RowTypeInfo and RowCompator in core #2968

Conversation

wuchong commented Dec 8, 2016 • edited

twalthr commented Dec 8, 2016

tonycox 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

tonycox Dec 8, 2016 • edited

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

wuchong commented Dec 9, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tonycox Dec 9, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tonycox Dec 9, 2016 • edited

Choose a reason for hiding this comment

tonycox Dec 12, 2016 • edited

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 12, 2016

wuchong commented Dec 12, 2016

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

Choose a reason for hiding this comment

fhueske commented Dec 12, 2016

wuchong commented Dec 12, 2016

tonycox commented Dec 12, 2016

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fhueske commented Dec 14, 2016

fhueske commented Dec 15, 2016

wuchong commented Dec 8, 2016 •

edited

tonycox Dec 8, 2016 •

edited

tonycox Dec 9, 2016 •

edited

tonycox Dec 9, 2016 •

edited

tonycox Dec 12, 2016 •

edited