You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by Shiti <gi...@git.apache.org> on 2015/06/12 21:04:30 UTC
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
GitHub user Shiti opened a pull request:
https://github.com/apache/flink/pull/831
[FLINK-2203] handling null values for RowSerializer
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/Shiti/flink FLINK-2203
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/831.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #831
----
commit 062e18e7b41c12389950ef4c84b49499c26254d1
Author: Shiti <ss...@gmail.com>
Date: 2015-06-12T15:25:48Z
[FLINK-2203] handling null values for RowSerializer
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/831#issuecomment-111717366
I had some more remarks, sorry for being so picky. :sweat_smile:
Other than that, I think the change looks really good now!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/831#discussion_r32371124
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowSerializer.scala ---
@@ -102,20 +119,39 @@ class RowSerializer(fieldSerializers: Array[TypeSerializer[Any]])
val len = fieldSerializers.length
val result = new Row(len)
- var i = 0
- while (i < len) {
- result.setField(i, fieldSerializers(i).deserialize(source))
- i += 1
+
+ var index = 0
+ while (index < len) {
+ val isNull: Boolean = source.readBoolean()
+ if (isNull) {
+ result.setField(index, null)
+ } else {
+ val serializer: TypeSerializer[Any] = fieldSerializers(index)
+ result.setField(index, serializer.deserialize(source))
+ }
+ index += 1
}
result
}
+ private final val booleanSerializer = new BooleanSerializer()
+
override def copy(source: DataInputView, target: DataOutputView): Unit = {
val len = fieldSerializers.length
var i = 0
while (i < len) {
+ booleanSerializer.copy(source, target)
--- End diff --
I think it would be easier to do
```
target.writeBoolean(source.readBoolean())
```
here, instead of going through the extra abstraction of the BooleanSerializer.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/831#issuecomment-111741260
Thanks, nice work. :+1:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by Shiti <gi...@git.apache.org>.
Github user Shiti commented on the pull request:
https://github.com/apache/flink/pull/831#issuecomment-111723371
@aljoscha I have made the suggested changes
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/831#discussion_r32369781
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowSerializer.scala ---
@@ -74,11 +79,16 @@ class RowSerializer(fieldSerializers: Array[TypeSerializer[Any]])
override def serialize(value: Row, target: DataOutputView) {
val len = fieldSerializers.length
- var i = 0
- while (i < len) {
- val serializer = fieldSerializers(i)
- serializer.serialize(value.productElement(i), target)
- i += 1
+ (0 to len - 1).foreach {
+ index =>
+ val o: AnyRef = value.productElement(index).asInstanceOf[AnyRef]
+ if (o == null) {
+ target.writeBoolean(true)
+ } else {
+ target.writeBoolean(false)
+ val serializer = fieldSerializers(index)
+ serializer.serialize(value.productElement(index), target)
+ }
--- End diff --
Could you please change this back to the simple while loop. I know that using the fancy Scala features is tempting but the performance of a simple while loop should be better than creating a range and iterating over it using foreach.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/831#issuecomment-111701489
The changes look good except for comments I had about the loops.
For the tests, did you try doing it as in TraversableSerializerTest.scala. Here, we override deepEquals() and the runTests() method can be modified to take the RowTypeInfo that was explicitly created. The reason why I would like it to derive from the SerializerTestBase as all the other serializers is that we have to change the tests for RowSerializer if we adapt the tests there. And this would likely be forgotten by future contributors.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/831#discussion_r32371128
--- Diff: flink-staging/flink-table/src/test/scala/org/apache/flink/api/table/typeinfo/RowSerializerTest.scala ---
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.typeinfo
+
+import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
+import org.apache.flink.api.common.typeutils.{SerializerTestInstance, TypeSerializer}
+import org.apache.flink.api.table.Row
+import org.junit.Assert._
+import org.junit.Test
+
+class RowSerializerTest {
+
+ class RowSerializerTestInstance(serializer: TypeSerializer[Row],
+ testData: Array[Row])
+ extends SerializerTestInstance(serializer, classOf[Row], -1, testData: _*) {
+
+ override protected def deepEquals(message: String, should: Row, is: Row): Unit = {
+ val arity = should.productArity
+ assertEquals(message, arity, is.productArity)
+ var index = 0
+ while (index < arity) {
+ val copiedValue: Any = should.productElement(index)
+ val element: Any = is.productElement(index)
+ assertEquals(message, element, copiedValue)
+ index += 1
+ }
+ }
+ }
+
+ @Test
+ def testRowSerializer(): Unit ={
+
+ val rowInfo: TypeInformation[Row] = new RowTypeInfo(
+ Seq(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO), Seq("id", "name"))
+
+ val row1 = new Row(2)
+ row1.setField(0, 1)
+ row1.setField(1, "a")
+
+ val row2 = new Row(2)
+ row2.setField(0, 2)
+ row2.setField(1, "hello")
+
+ val testData: Array[Row] = Array(row1, row2)
--- End diff --
I think it would be good to also add a row that has null values, since the change actually introduces that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by Shiti <gi...@git.apache.org>.
Github user Shiti commented on the pull request:
https://github.com/apache/flink/pull/831#issuecomment-111709239
@aljoscha I have updated the RowSerializerTest to use SerializerTestBase and reverted to using while loops in RowSerializer
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by Shiti <gi...@git.apache.org>.
Github user Shiti commented on the pull request:
https://github.com/apache/flink/pull/831#issuecomment-111682105
@hsaputra I have updated the description
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by hsaputra <gi...@git.apache.org>.
Github user hsaputra commented on the pull request:
https://github.com/apache/flink/pull/831#issuecomment-111627080
HI @Shiti , thanks for the patch.
Could you add in the PR description on how the fix is done and what changes made to fix it.
This will help reviewers follow your line of thoughts for the fix.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/831
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2203] handling null values for RowSeria...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/831#discussion_r32369782
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RowSerializer.scala ---
@@ -89,11 +99,16 @@ class RowSerializer(fieldSerializers: Array[TypeSerializer[Any]])
throw new RuntimeException("Row arity of reuse and fields do not match.")
}
- var i = 0
- while (i < len) {
- val field = reuse.productElement(i).asInstanceOf[AnyRef]
- reuse.setField(i, fieldSerializers(i).deserialize(field, source))
- i += 1
+ (0 to len - 1).foreach {
+ index =>
+ val isNull: Boolean = source.readBoolean
+ if (isNull) {
+ reuse.setField(index, null)
+ } else {
+ val field = reuse.productElement(index).asInstanceOf[AnyRef]
+ val serializer: TypeSerializer[Any] = fieldSerializers(index)
+ reuse.setField(index, serializer.deserialize(field, source))
+ }
--- End diff --
See above comment about while loop.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---