Repository: crunch Updated Branches: refs/heads/master ccad69448 -> 86ecd82d9
CRUNCH-578: Add support for mutable collection type serialization to Scrunch Project: http://git-wip-us.apache.org/repos/asf/crunch/repo Commit: http://git-wip-us.apache.org/repos/asf/crunch/commit/86ecd82d Tree: http://git-wip-us.apache.org/repos/asf/crunch/tree/86ecd82d Diff: http://git-wip-us.apache.org/repos/asf/crunch/diff/86ecd82d Branch: refs/heads/master Commit: 86ecd82d9a8510ebb9254a24cfba594efcce4ad2 Parents: ccad694 Author: Josh Wills <[email protected]> Authored: Mon Nov 16 17:35:48 2015 -0800 Committer: Josh Wills <[email protected]> Committed: Mon Nov 16 17:35:48 2015 -0800 ---------------------------------------------------------------------- .../org/apache/crunch/scrunch/Conversions.scala | 25 ++++++++++++++++++++ .../org/apache/crunch/scrunch/PTypeFamily.scala | 20 ++++++++++++++++ .../org/apache/crunch/scrunch/TupleNTest.scala | 8 +++++++ 3 files changed, 53 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/crunch/blob/86ecd82d/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/Conversions.scala ---------------------------------------------------------------------- diff --git a/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/Conversions.scala b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/Conversions.scala index 5807490..a108a44 100644 --- a/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/Conversions.scala +++ b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/Conversions.scala @@ -22,6 +22,7 @@ import org.apache.crunch.{Pair => CPair} import org.apache.crunch.types.{PTypes, PType} import java.nio.ByteBuffer import scala.collection.Iterable +import scala.collection.mutable.{ListBuffer, Set => MSet, Map => MMap} import scala.reflect.ClassTag import scala.reflect.runtime.universe.TypeTag import org.apache.hadoop.io.Writable @@ -178,6 +179,14 @@ object PTypeH extends GeneratedTupleConversions with LowPriorityPTypeH { } } + implicit def listbuffers[T: PTypeH] = { + new PTypeH[ListBuffer[T]] { + def get(ptf: PTypeFamily) = { + ptf.listbuffers(implicitly[PTypeH[T]].get(ptf)) + } + } + } + implicit def sets[T: PTypeH] = { new PTypeH[Set[T]] { def get(ptf: PTypeFamily) = { @@ -186,6 +195,14 @@ object PTypeH extends GeneratedTupleConversions with LowPriorityPTypeH { } } + implicit def msets[T: PTypeH] = { + new PTypeH[MSet[T]] { + def get(ptf: PTypeFamily) = { + ptf.mutableSets(implicitly[PTypeH[T]].get(ptf)) + } + } + } + implicit def maps[K: PTypeH, V: PTypeH] = { new PTypeH[Map[K, V]] { def get(ptf: PTypeFamily) = { @@ -194,6 +211,14 @@ object PTypeH extends GeneratedTupleConversions with LowPriorityPTypeH { } } + implicit def mmaps[K: PTypeH, V: PTypeH] = { + new PTypeH[MMap[K, V]] { + def get(ptf: PTypeFamily) = { + ptf.mutableMaps(implicitly[PTypeH[K]].get(ptf), implicitly[PTypeH[V]].get(ptf)) + } + } + } + implicit def pairs[A: PTypeH, B: PTypeH] = { new PTypeH[(A, B)] { def get(ptf: PTypeFamily) = { http://git-wip-us.apache.org/repos/asf/crunch/blob/86ecd82d/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PTypeFamily.scala ---------------------------------------------------------------------- diff --git a/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PTypeFamily.scala b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PTypeFamily.scala index e9992ef..47cf637 100644 --- a/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PTypeFamily.scala +++ b/crunch-scrunch/src/main/scala/org/apache/crunch/scrunch/PTypeFamily.scala @@ -25,6 +25,7 @@ import java.lang.{Long => JLong, Double => JDouble, Integer => JInt, Float => JF import java.lang.reflect.{Array => RArray} import java.util.{Collection => JCollection} import scala.collection.JavaConversions._ +import scala.collection.mutable.{ListBuffer, Set => MSet, Map => MMap} import scala.reflect.ClassTag import scala.reflect.runtime.universe._ import scala.reflect.runtime.currentMirror @@ -187,6 +188,13 @@ trait PTypeFamily extends GeneratedTuplePTypeFamily { } } + def mutableMaps[K, V](keyType: PType[K], valueType: PType[V]): PType[MMap[K, V]] = { + derived(classOf[MMap[K, V]], + {x: JCollection[CPair[K, V]] => MMap[K, V](x.map(y => (y.first(), y.second())).toArray : _*)}, + {x: MMap[K, V] => asJavaCollection(x.toIterable.map(y => CPair.of(y._1, y._2)))}, + ptf.collections(ptf.pairs(keyType, valueType))) + } + def arrays[T](ptype: PType[T]): PType[Array[T]] = { val in = (x: JCollection[_]) => { val ret = RArray.newInstance(ptype.getTypeClass, x.size()) @@ -211,12 +219,24 @@ trait PTypeFamily extends GeneratedTuplePTypeFamily { derived(classOf[List[T]], in, out, ptf.collections(ptype)) } + def listbuffers[T](ptype: PType[T]) = { + val in = (x: JCollection[T]) => collectionAsScalaIterable[T](x).to[ListBuffer] + val out = (x: ListBuffer[T]) => asJavaCollection[T](x) + derived(classOf[ListBuffer[T]], in, out, ptf.collections(ptype)) + } + def sets[T](ptype: PType[T]) = { val in = (x: JCollection[T]) => collectionAsScalaIterable[T](x).toSet val out = (x: Set[T]) => asJavaCollection[T](x) derived(classOf[Set[T]], in, out, ptf.collections(ptype)) } + def mutableSets[T](ptype: PType[T]) = { + val in = (x: JCollection[T]) => collectionAsScalaIterable[T](x).to[MSet] + val out = (x: MSet[T]) => asJavaCollection[T](x) + derived(classOf[MSet[T]], in, out, ptf.collections(ptype)) + } + def tuple2[T1, T2](p1: PType[T1], p2: PType[T2]) = { val in = (x: CPair[T1, T2]) => (x.first(), x.second()) val out = (x: (T1, T2)) => CPair.of(x._1, x._2) http://git-wip-us.apache.org/repos/asf/crunch/blob/86ecd82d/crunch-scrunch/src/test/scala/org/apache/crunch/scrunch/TupleNTest.scala ---------------------------------------------------------------------- diff --git a/crunch-scrunch/src/test/scala/org/apache/crunch/scrunch/TupleNTest.scala b/crunch-scrunch/src/test/scala/org/apache/crunch/scrunch/TupleNTest.scala index 50e92ed..8105217 100644 --- a/crunch-scrunch/src/test/scala/org/apache/crunch/scrunch/TupleNTest.scala +++ b/crunch-scrunch/src/test/scala/org/apache/crunch/scrunch/TupleNTest.scala @@ -25,6 +25,8 @@ import org.apache.crunch.types.avro.AvroMode import org.scalatest.junit.JUnitSuite import org.junit.Test +import scala.collection.mutable.{ListBuffer, Set => MSet, Map => MMap} + /** Case classes for testing purposes */ case class One(a: Int, b: String, c: List[java.lang.Long], d: Array[Long]) case class Two(a: One, b: Set[Option[Boolean]], c: Map[String, Double], d: Map[Int, String]) @@ -43,6 +45,12 @@ class TupleNTest extends JUnitSuite{ org.junit.Assert.assertEquals(List((AvroMode.SPECIFIC, 2)), res.toList) } + @Test def testMutablePTypes { + val lb = Mem.collectionOf(ListBuffer[Int](1, 2, 3), ListBuffer[Int](4, 5, 6)) + val ms = Mem.collectionOf(MSet[String]("a", "b", "b"), MSet[String]("x", "y", "x")) + val mm = Mem.collectionOf(MMap[Int, String](1 -> "a", 2 -> "b", 3 -> "c"), MMap[Int, String]()) + } + /** * Basically, we just want to validate that we can generate schemas for these classes successfully */
