You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/01/11 22:51:39 UTC
[1/3] flink git commit: [FLINK-1378] [scala] Fix type extraction for
nested type parameters
Repository: flink
Updated Branches:
refs/heads/release-0.8 3da8751c6 -> ada35eb05
[FLINK-1378] [scala] Fix type extraction for nested type parameters
Before, something like:
def f[T: TypeInformation](data: DataSet[T]) = {
val tpe = createTypeInformation[(T, Seq[T])]
println("Type: " + tpe)
}
f(Seq(1.0f, 2.0f)
would fail because the type extractor could not re-use existing
TypeInformation for nested types.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4f9dcaee
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4f9dcaee
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4f9dcaee
Branch: refs/heads/release-0.8
Commit: 4f9dcaeea37fc95042a62fa63366bfb71439a563
Parents: 3da8751
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Fri Jan 9 17:11:17 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Sun Jan 11 21:46:20 2015 +0100
----------------------------------------------------------------------
.../flink/api/scala/codegen/TypeAnalyzer.scala | 8 +++++++-
.../api/scala/codegen/TypeDescriptors.scala | 6 ++++++
.../api/scala/codegen/TypeInformationGen.scala | 20 +++++++++++++++++++
.../scala/types/TypeInformationGenTest.scala | 21 ++++++++++++++++++++
4 files changed, 54 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4f9dcaee/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
index 64b9543..2ab332a 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
@@ -31,7 +31,6 @@ import org.apache.flink.types.StringValue
import org.apache.flink.types.LongValue
import org.apache.flink.types.ShortValue
-
private[flink] trait TypeAnalyzer[C <: Context] { this: MacroContextHolder[C]
with TypeDescriptors[C] =>
@@ -54,6 +53,9 @@ private[flink] trait TypeAnalyzer[C <: Context] { this: MacroContextHolder[C]
cache.getOrElseUpdate(tpe) { id =>
tpe match {
+
+ case TypeParameter() => TypeParameterDescriptor(id, tpe)
+
case PrimitiveType(default, wrapper) => PrimitiveDescriptor(id, tpe, default, wrapper)
case BoxedPrimitiveType(default, wrapper, box, unbox) =>
@@ -282,6 +284,10 @@ private[flink] trait TypeAnalyzer[C <: Context] { this: MacroContextHolder[C]
}
}
+ private object TypeParameter {
+ def unapply(tpe: Type): Boolean = tpe.typeSymbol.isParameter
+ }
+
private object CaseClassType {
def unapply(tpe: Type): Boolean = tpe.typeSymbol.asClass.isCaseClass
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4f9dcaee/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
index f96cde5..f013abb 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
@@ -73,6 +73,12 @@ private[flink] trait TypeDescriptors[C <: Context] { this: MacroContextHolder[C]
def canBeKey = tpe <:< typeOf[Comparable[_]]
}
+ case class TypeParameterDescriptor(id: Int, tpe: Type) extends UDTDescriptor {
+ override val isPrimitiveProduct = false
+ override def flatten = Seq(this)
+ override def canBeKey = false
+ }
+
case class PrimitiveDescriptor(id: Int, tpe: Type, default: Literal, wrapper: Type)
extends UDTDescriptor {
override val isPrimitiveProduct = true
http://git-wip-us.apache.org/repos/asf/flink/blob/4f9dcaee/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
index 784c5fd..f1b2400 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
@@ -54,6 +54,8 @@ private[flink] trait TypeInformationGen[C <: Context] {
mkCaseClassTypeInfo(cc)(c.WeakTypeTag(tpe).asInstanceOf[c.WeakTypeTag[Product]])
.asInstanceOf[c.Expr[TypeInformation[T]]]
+ case tp: TypeParameterDescriptor => mkTypeParameter(tp)
+
case p : PrimitiveDescriptor => mkPrimitiveTypeInfo(p.tpe)
case p : BoxedPrimitiveDescriptor => mkPrimitiveTypeInfo(p.tpe)
@@ -271,6 +273,24 @@ private[flink] trait TypeInformationGen[C <: Context] {
}
}
+ def mkTypeParameter[T: c.WeakTypeTag](
+ typeParameter: TypeParameterDescriptor): c.Expr[TypeInformation[T]] = {
+
+ val result = c.inferImplicitValue(
+ c.weakTypeOf[TypeInformation[T]],
+ silent = true,
+ withMacrosDisabled = false,
+ pos = c.enclosingPosition)
+
+ if (result.isEmpty) {
+ c.error(
+ c.enclosingPosition,
+ s"could not find implicit value of type TypeInformation[${typeParameter.tpe}].")
+ }
+
+ c.Expr[TypeInformation[T]](result)
+ }
+
def mkPrimitiveTypeInfo[T: c.WeakTypeTag](tpe: Type): c.Expr[TypeInformation[T]] = {
val tpeClazz = c.Expr[Class[T]](Literal(Constant(tpe)))
reify {
http://git-wip-us.apache.org/repos/asf/flink/blob/4f9dcaee/flink-tests/src/test/scala/org/apache/flink/api/scala/types/TypeInformationGenTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/types/TypeInformationGenTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/types/TypeInformationGenTest.scala
index 2b2d3a9..53df802 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/types/TypeInformationGenTest.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/types/TypeInformationGenTest.scala
@@ -21,6 +21,7 @@ import java.io.DataInput
import java.io.DataOutput
import org.apache.flink.api.common.typeinfo._
import org.apache.flink.api.java.typeutils._
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
import org.apache.flink.types.{IntValue, StringValue}
import org.apache.hadoop.io.Writable
import org.junit.Assert
@@ -60,6 +61,26 @@ class TypeInformationGenTest {
}
@Test
+ def testTypeParameters(): Unit = {
+
+ val data = Seq(1.0d, 2.0d)
+
+ def f[T: TypeInformation](data: Seq[T]): (T, Seq[T]) = {
+
+ val ti = createTypeInformation[(T, Seq[T])]
+
+ Assert.assertTrue(ti.isTupleType)
+ val ccti = ti.asInstanceOf[CaseClassTypeInfo[(T, Seq[T])]]
+ Assert.assertEquals(BasicTypeInfo.DOUBLE_TYPE_INFO, ccti.getTypeAt(0))
+
+ (data.head, data)
+ }
+
+ f(data)
+
+ }
+
+ @Test
def testWritableType(): Unit = {
val ti = createTypeInformation[MyWritable]
[3/3] flink git commit: [FLINK-1378] [scala] Add support for Try[A]
(Success/Failure)
Posted by se...@apache.org.
[FLINK-1378] [scala] Add support for Try[A] (Success/Failure)
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ada35eb0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ada35eb0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ada35eb0
Branch: refs/heads/release-0.8
Commit: ada35eb059400c68d8e289cd98ca024c5ac87dcd
Parents: fced2eb
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Fri Jan 9 19:31:05 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Sun Jan 11 22:24:38 2015 +0100
----------------------------------------------------------------------
.../flink/api/scala/codegen/TypeAnalyzer.scala | 24 ++++++
.../api/scala/codegen/TypeDescriptors.scala | 7 ++
.../api/scala/codegen/TypeInformationGen.scala | 15 ++++
.../api/scala/typeutils/OptionTypeInfo.scala | 4 +-
.../api/scala/typeutils/TrySerializer.scala | 89 ++++++++++++++++++++
.../flink/api/scala/typeutils/TryTypeInfo.scala | 49 +++++++++++
.../scala/runtime/ScalaSpecialTypesITCase.scala | 60 ++++++++++++-
.../ScalaSpecialTypesSerializerTest.scala | 33 +++++++-
8 files changed, 272 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
index 2ab332a..12efd21 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeAnalyzer.scala
@@ -69,6 +69,8 @@ private[flink] trait TypeAnalyzer[C <: Context] { this: MacroContextHolder[C]
case EitherType(leftTpe, rightTpe) => analyzeEither(id, tpe, leftTpe, rightTpe)
+ case TryType(elemTpe) => analyzeTry(id, tpe, elemTpe)
+
case OptionType(elemTpe) => analyzeOption(id, tpe, elemTpe)
case CaseClassType() => analyzeCaseClass(id, tpe)
@@ -116,6 +118,14 @@ private[flink] trait TypeAnalyzer[C <: Context] { this: MacroContextHolder[C]
}
}
+ private def analyzeTry(
+ id: Int,
+ tpe: Type,
+ elemTpe: Type): UDTDescriptor = analyze(elemTpe) match {
+ case UnsupportedDescriptor(_, _, errs) => UnsupportedDescriptor(id, tpe, errs)
+ case elemDesc => TryDescriptor(id, tpe, elemDesc)
+ }
+
private def analyzeOption(
id: Int,
tpe: Type,
@@ -310,6 +320,20 @@ private[flink] trait TypeAnalyzer[C <: Context] { this: MacroContextHolder[C]
}
}
+ private object TryType {
+ def unapply(tpe: Type): Option[Type] = {
+ if (tpe <:< typeOf[scala.util.Try[_]]) {
+ val option = tpe.baseType(typeOf[scala.util.Try[_]].typeSymbol)
+ option match {
+ case TypeRef(_, _, elemTpe :: Nil) =>
+ Some(elemTpe)
+ }
+ } else {
+ None
+ }
+ }
+ }
+
private object OptionType {
def unapply(tpe: Type): Option[Type] = {
if (tpe <:< typeOf[Option[_]]) {
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
index f013abb..71cfe7c 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeDescriptors.scala
@@ -100,6 +100,13 @@ private[flink] trait TypeDescriptors[C <: Context] { this: MacroContextHolder[C]
override def canBeKey = false
}
+ case class TryDescriptor(id: Int, tpe: Type, elem: UDTDescriptor)
+ extends UDTDescriptor {
+ override val isPrimitiveProduct = false
+ override def flatten = Seq(this)
+ override def canBeKey = false
+ }
+
case class OptionDescriptor(id: Int, tpe: Type, elem: UDTDescriptor)
extends UDTDescriptor {
override val isPrimitiveProduct = false
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
index f1b2400..8e80132 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
@@ -63,6 +63,8 @@ private[flink] trait TypeInformationGen[C <: Context] {
case e: EitherDescriptor => mkEitherTypeInfo(e)
+ case tr: TryDescriptor => mkTryTypeInfo(tr)
+
case o: OptionDescriptor => mkOptionTypeInfo(o)
case a : ArrayDescriptor => mkArrayTypeInfo(a)
@@ -129,6 +131,19 @@ private[flink] trait TypeInformationGen[C <: Context] {
c.Expr[TypeInformation[T]](result)
}
+ def mkTryTypeInfo[T: c.WeakTypeTag](desc: TryDescriptor): c.Expr[TypeInformation[T]] = {
+
+ val elemTypeInfo = mkTypeInfo(desc.elem)(c.WeakTypeTag(desc.elem.tpe))
+
+ val result = q"""
+ import org.apache.flink.api.scala.typeutils.TryTypeInfo
+
+ new TryTypeInfo[${desc.elem.tpe}, ${desc.tpe}]($elemTypeInfo)
+ """
+
+ c.Expr[TypeInformation[T]](result)
+ }
+
def mkOptionTypeInfo[T: c.WeakTypeTag](desc: OptionDescriptor): c.Expr[TypeInformation[T]] = {
val elemTypeInfo = mkTypeInfo(desc.elem)(c.WeakTypeTag(desc.elem.tpe))
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionTypeInfo.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionTypeInfo.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionTypeInfo.scala
index 171db60..efa2ab8 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionTypeInfo.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionTypeInfo.scala
@@ -21,7 +21,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.common.typeutils.TypeSerializer
/**
- * TypeInformation [[Option]].
+ * TypeInformation for [[Option]].
*/
class OptionTypeInfo[A, T <: Option[A]](elemTypeInfo: TypeInformation[A])
extends TypeInformation[T] {
@@ -35,7 +35,7 @@ class OptionTypeInfo[A, T <: Option[A]](elemTypeInfo: TypeInformation[A])
def createSerializer(): TypeSerializer[T] = {
if (elemTypeInfo == null) {
- // this happens when the type of a DataSet is None
+ // this happens when the type of a DataSet is None, i.e. DataSet[None]
new OptionSerializer(new NothingSerializer).asInstanceOf[TypeSerializer[T]]
} else {
new OptionSerializer(elemTypeInfo.createSerializer()).asInstanceOf[TypeSerializer[T]]
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala
new file mode 100644
index 0000000..530ca7a
--- /dev/null
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TrySerializer.scala
@@ -0,0 +1,89 @@
+/*
+ * 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.scala.typeutils
+
+import org.apache.flink.api.common.typeutils.TypeSerializer
+import org.apache.flink.api.java.typeutils.runtime.KryoSerializer
+import org.apache.flink.core.memory.{DataInputView, DataOutputView}
+
+import scala.util.{Success, Try, Failure}
+
+/**
+ * Serializer for [[scala.util.Try]].
+ */
+class TrySerializer[A](val elemSerializer: TypeSerializer[A])
+ extends TypeSerializer[Try[A]] {
+
+ override def isStateful: Boolean = false
+
+ val throwableSerializer = new KryoSerializer[Throwable](classOf[Throwable])
+
+ override def createInstance: Try[A] = {
+ Failure(new RuntimeException("Empty Failure"))
+ }
+
+ override def isImmutableType: Boolean = elemSerializer == null || elemSerializer.isImmutableType
+
+ override def getLength: Int = -1
+
+ override def copy(from: Try[A]): Try[A] = from match {
+ case Success(a) => Success(elemSerializer.copy(a))
+ case Failure(t) => Failure(throwableSerializer.copy(t))
+ }
+
+ override def copy(from: Try[A], reuse: Try[A]): Try[A] = copy(from)
+
+ override def copy(source: DataInputView, target: DataOutputView): Unit = {
+ val isSuccess = source.readBoolean()
+ target.writeBoolean(isSuccess)
+ if (isSuccess) {
+ elemSerializer.copy(source, target)
+ } else {
+ throwableSerializer.copy(source, target)
+ }
+ }
+
+ override def serialize(either: Try[A], target: DataOutputView): Unit = either match {
+ case Success(a) =>
+ target.writeBoolean(true)
+ elemSerializer.serialize(a, target)
+ case Failure(t) =>
+ target.writeBoolean(false)
+ throwableSerializer.serialize(t, target)
+ }
+
+ override def deserialize(source: DataInputView): Try[A] = {
+ val isSuccess = source.readBoolean()
+ if (isSuccess) {
+ Success(elemSerializer.deserialize(source))
+ } else {
+ Failure(throwableSerializer.deserialize(source))
+ }
+ }
+
+ override def deserialize(reuse: Try[A], source: DataInputView): Try[A] = deserialize(source)
+
+ override def equals(obj: Any): Boolean = {
+ if (obj != null && obj.isInstanceOf[TrySerializer[_]]) {
+ val other = obj.asInstanceOf[TrySerializer[_]]
+ other.elemSerializer.equals(elemSerializer)
+ } else {
+ false
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TryTypeInfo.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TryTypeInfo.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TryTypeInfo.scala
new file mode 100644
index 0000000..f433eea
--- /dev/null
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/TryTypeInfo.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.scala.typeutils
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeutils.TypeSerializer
+import org.apache.flink.api.java.typeutils.runtime.KryoSerializer
+
+import scala.util.Try
+
+/**
+ * TypeInformation for [[scala.util.Try]].
+ */
+class TryTypeInfo[A, T <: Try[A]](elemTypeInfo: TypeInformation[A])
+ extends TypeInformation[T] {
+
+ override def isBasicType: Boolean = false
+ override def isTupleType: Boolean = false
+ override def isKeyType: Boolean = false
+ override def getTotalFields: Int = 1
+ override def getArity: Int = 1
+ override def getTypeClass = classOf[Try[_]].asInstanceOf[Class[T]]
+
+ def createSerializer(): TypeSerializer[T] = {
+ if (elemTypeInfo == null) {
+ // this happens when the type of a DataSet is None, i.e. DataSet[Failure]
+ new TrySerializer(new NothingSerializer).asInstanceOf[TypeSerializer[T]]
+ } else {
+ new TrySerializer(elemTypeInfo.createSerializer()).asInstanceOf[TypeSerializer[T]]
+ }
+ }
+
+ override def toString = s"Try[$elemTypeInfo]"
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
index a923af6..ebed7e7 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesITCase.scala
@@ -25,11 +25,11 @@ import org.apache.flink.test.util.JavaProgramTestBase
import org.junit.runner.RunWith
import org.junit.runners.Parameterized
import org.junit.runners.Parameterized.Parameters
-
import scala.collection.JavaConverters._
import scala.collection.mutable
-
import org.apache.flink.api.scala._
+import scala.util.Success
+import scala.util.Failure
@RunWith(classOf[Parameterized])
@@ -149,6 +149,60 @@ class ScalaSpecialTypesITCase(config: Configuration) extends JavaProgramTestBase
"80"
+ case 7 =>
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val nums = env.fromElements(1, 2, 1, 2)
+
+ val trys = nums.map(_ match {
+ case 1 => Success(10)
+ case 2 => Failure(new RuntimeException("20"))
+ })
+
+ val result = trys.map{
+ _ match {
+ case Success(i) => i
+ case Failure(t) => t.getMessage.toInt
+ }}.reduce(_ + _).writeAsText(resultPath)
+
+ env.execute()
+
+ "60"
+
+ case 8 =>
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val nums = env.fromElements(1, 2, 1, 2)
+
+ val trys = nums.map(_ match {
+ case 1 => Success(10)
+ case 2 => Success(20)
+ })
+
+
+ val result = trys.map(_ match {
+ case Success(i) => i
+ }).reduce(_ + _).writeAsText(resultPath)
+
+ env.execute()
+
+ "60"
+
+ case 9 =>
+ val env = ExecutionEnvironment.getExecutionEnvironment
+ val nums = env.fromElements(1, 2, 1, 2)
+
+ val trys = nums.map(_ match {
+ case 1 => Failure(new RuntimeException("10"))
+ case 2 => Failure(new IllegalAccessError("20"))
+ })
+
+ val result = trys.map(_ match {
+ case Failure(t) => t.getMessage.toInt
+ }).reduce(_ + _).writeAsText(resultPath)
+
+ env.execute()
+
+ "60"
+
case _ =>
throw new IllegalArgumentException("Invalid program id")
}
@@ -160,7 +214,7 @@ class ScalaSpecialTypesITCase(config: Configuration) extends JavaProgramTestBase
}
object ScalaSpecialTypesITCase {
- var NUM_PROGRAMS: Int = 6
+ var NUM_PROGRAMS: Int = 9
@Parameters
def getConfigurations: java.util.Collection[Array[AnyRef]] = {
http://git-wip-us.apache.org/repos/asf/flink/blob/ada35eb0/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesSerializerTest.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesSerializerTest.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesSerializerTest.scala
index 60651d1..297ecc5 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesSerializerTest.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/ScalaSpecialTypesSerializerTest.scala
@@ -25,6 +25,8 @@ import org.junit.{Assert, Test}
import org.apache.flink.api.scala._
+import scala.util.{Failure, Success}
+
class ScalaSpecialTypesSerializerTest {
@Test
@@ -63,6 +65,26 @@ class ScalaSpecialTypesSerializerTest {
runTests(testData)
}
+ @Test
+ def testTry(): Unit = {
+ val testData = Array(Success("Hell"), Failure(new RuntimeException("test")))
+ runTests(testData)
+ }
+
+ @Test
+ def testSuccess(): Unit = {
+ val testData = Array(Success("Hell"), Success("Yeah"))
+ runTests(testData)
+ }
+
+ @Test
+ def testFailure(): Unit = {
+ val testData = Array(
+ Failure(new RuntimeException("test")),
+ Failure(new RuntimeException("one, two")))
+ runTests(testData)
+ }
+
private final def runTests[T : TypeInformation](instances: Array[T]) {
try {
@@ -83,10 +105,10 @@ class ScalaSpecialTypesSerializerTest {
}
class ScalaSpecialTypesSerializerTestInstance[T](
- serializer: TypeSerializer[T],
- typeClass: Class[T],
- length: Int,
- testData: Array[T])
+ serializer: TypeSerializer[T],
+ typeClass: Class[T],
+ length: Int,
+ testData: Array[T])
extends SerializerTestInstance[T](serializer, typeClass, length, testData: _*) {
@Test
@@ -123,6 +145,9 @@ class ScalaSpecialTypesSerializerTestInstance[T](
assertEquals(message, should, is)
}
+ case Failure(t) =>
+ is.asInstanceOf[Failure[_]].exception.equals(t)
+
case _ =>
super.deepEquals(message, should, is)
}
[2/3] flink git commit: [FLINK-1378] Add support for Throwables in
KryoSerializer
Posted by se...@apache.org.
[FLINK-1378] Add support for Throwables in KryoSerializer
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fced2eb4
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fced2eb4
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fced2eb4
Branch: refs/heads/release-0.8
Commit: fced2eb4f36ee59363c1ba1331b4ab457c30fe89
Parents: 4f9dcae
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Fri Jan 9 19:04:23 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Sun Jan 11 21:48:03 2015 +0100
----------------------------------------------------------------------
.../flink/api/common/typeutils/SerializerTestBase.java | 3 +++
.../api/java/typeutils/runtime/KryoSerializer.java | 13 +++++++++----
.../scala/runtime/KryoGenericTypeSerializerTest.scala | 13 ++++++++++---
3 files changed, 22 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fced2eb4/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java
index 5122af9..4835b4f 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java
@@ -384,6 +384,9 @@ public abstract class SerializerTestBase<T> {
assertArrayEquals(message, (Object[]) should, (Object[]) is);
}
}
+ else if (should instanceof Throwable) {
+ assertEquals(((Throwable)should).getMessage(), ((Throwable)is).getMessage());
+ }
else {
assertEquals(message, should, is);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/fced2eb4/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
index 186390a..f445ec6 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
@@ -22,8 +22,9 @@ import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoException;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
-
+import com.esotericsoftware.kryo.serializers.JavaSerializer;
import com.twitter.chill.ScalaKryoInstantiator;
+
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
@@ -70,22 +71,24 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
return null;
}
+ @SuppressWarnings("unchecked")
@Override
public T copy(T from) {
- if(from == null) {
+ if (from == null) {
return null;
}
checkKryoInitialized();
try {
return kryo.copy(from);
- } catch(KryoException ke) {
+ }
+ catch(KryoException ke) {
// kryo was unable to copy it, so we do it through serialization:
ByteArrayOutputStream baout = new ByteArrayOutputStream();
Output output = new Output(baout);
kryo.writeObject(output, from);
- output.flush();
+ output.close();
ByteArrayInputStream bain = new ByteArrayInputStream(baout.toByteArray());
Input input = new Input(bain);
@@ -128,6 +131,7 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
}
}
+ @SuppressWarnings("unchecked")
@Override
public T deserialize(DataInputView source) throws IOException {
checkKryoInitialized();
@@ -177,6 +181,7 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
private void checkKryoInitialized() {
if (this.kryo == null) {
this.kryo = new ScalaKryoInstantiator().newKryo();
+ this.kryo.addDefaultSerializer(Throwable.class, new JavaSerializer());
this.kryo.setRegistrationRequired(false);
this.kryo.register(type);
this.kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
http://git-wip-us.apache.org/repos/asf/flink/blob/fced2eb4/flink-scala/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
index ddbe322..e02ef5f 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/runtime/KryoGenericTypeSerializerTest.scala
@@ -26,6 +26,13 @@ import scala.reflect._
class KryoGenericTypeSerializerTest {
@Test
+ def testThrowableSerialization: Unit = {
+ val a = List(new RuntimeException("Hello"), new RuntimeException("there"))
+
+ runTests(a)
+ }
+
+ @Test
def testScalaListSerialization: Unit = {
val a = List(42,1,49,1337)
@@ -43,14 +50,14 @@ class KryoGenericTypeSerializerTest {
def testScalaMapSerialization: Unit = {
val a = Map(("1" -> 1), ("2" -> 2), ("42" -> 42), ("1337" -> 1337))
- runTests(a)
+ runTests(Seq(a))
}
@Test
def testMutableMapSerialization: Unit ={
val a = scala.collection.mutable.Map((1 -> "1"), (2 -> "2"), (3 -> "3"))
- runTests(a)
+ runTests(Seq(a))
}
@Test
@@ -115,7 +122,7 @@ class KryoGenericTypeSerializerTest {
}
}
- def runTests[T : ClassTag](objects: T *): Unit ={
+ def runTests[T : ClassTag](objects: Seq[T]): Unit ={
val clsTag = classTag[T]
val typeInfo = new GenericTypeInfo[T](clsTag.runtimeClass.asInstanceOf[Class[T]])
val serializer = typeInfo.createSerializer()