Add java priority queue, set, deque, collection coders

scio-core/src/main/scala/com/spotify/scio/coders/instances/JavaCoders.scala

@@ -20,7 +20,6 @@ package com.spotify.scio.coders.instances
 import java.io.{InputStream, OutputStream}
 import java.math.{BigDecimal, BigInteger}
 import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime, Period}
-import java.util.UUID
 import com.spotify.scio.IsJavaBean
 import com.spotify.scio.coders.{Coder, CoderGrammar}
 import com.spotify.scio.schemas.Schema
@@ -48,9 +47,9 @@ private[coders] object VoidCoder extends AtomicCoder[Void] {
 trait JavaCoders extends CoderGrammar with JavaBeanCoders {
   implicit lazy val voidCoder: Coder[Void] = beam[Void](VoidCoder)
 
-  implicit lazy val uuidCoder: Coder[UUID] =
+  implicit lazy val uuidCoder: Coder[java.util.UUID] =
     xmap(Coder[(Long, Long)])(
-      { case (msb, lsb) => new UUID(msb, lsb) },
+      { case (msb, lsb) => new java.util.UUID(msb, lsb) },
       uuid => (uuid.getMostSignificantBits, uuid.getLeastSignificantBits)
     )
 
@@ -69,6 +68,29 @@ trait JavaCoders extends CoderGrammar with JavaBeanCoders {
   implicit def jArrayListCoder[T](implicit c: Coder[T]): Coder[java.util.ArrayList[T]] =
     xmap(jListCoder[T])(new java.util.ArrayList(_), identity)
 
+  /**
+   * Coder must be created explicitly via
+   * {{{implicit val pqCoder = Coders.jPriorityQueueCoder[T](ord)}}} since there is otherwise no
+   * guarantee that the PriorityQueue comparator is the same as an implicitly-available ordering at
+   * coder construction time.
+   *
+   * @param ord
+   *   Ordering used as PriorityQueue comparator
+   */
+  def jPriorityQueueCoder[T: Coder: ClassTag](
+    ord: Ordering[T]
+  ): Coder[java.util.PriorityQueue[T]] = {
+    // neither arrays nor PriorityQueues are consistentWithEquals
+    Coder.xmap(ScalaCoders.arrayCoder[T])(
+      arr => {
+        val pq = new java.util.PriorityQueue[T](ord)
+        pq.addAll(java.util.Arrays.asList(arr: _*))
+        pq
+      },
+      _.toArray.asInstanceOf[Array[T]]
+    )
+  }
+
   implicit def jMapCoder[K, V](implicit ck: Coder[K], cv: Coder[V]): Coder[java.util.Map[K, V]] =
     transform(ck)(bk => transform(cv)(bv => beam(bcoders.MapCoder.of(bk, bv))))
 

scio-core/src/main/scala/com/spotify/scio/coders/instances/ScalaCoders.scala

@@ -248,7 +248,7 @@ private class SortedSetCoder[T: Ordering](bc: BCoder[T]) extends SeqLikeCoder[So
     decode(inStream, SortedSet.newBuilder[T])
 }
 
-private class MutablePriorityQueueCoder[T: Ordering](bc: BCoder[T])
+private[coders] class MutablePriorityQueueCoder[T: Ordering](bc: BCoder[T])
     extends SeqLikeCoder[m.PriorityQueue, T](bc) {
   override def consistentWithEquals(): Boolean = false // PriorityQueue does not define equality
   override def decode(inStream: InputStream): m.PriorityQueue[T] =

scio-core/src/test/scala/com/spotify/scio/coders/CoderTest.scala

@@ -33,6 +33,7 @@ import org.apache.beam.sdk.util.SerializableUtils
 import org.apache.beam.sdk.extensions.protobuf.ByteStringCoder
 import org.apache.beam.sdk.schemas.SchemaCoder
 import org.apache.commons.io.output.NullOutputStream
+import org.scalactic.Equality
 import org.scalatest.Assertion
 import org.scalatest.exceptions.TestFailedException
 import org.scalatest.flatspec.AnyFlatSpec
@@ -43,9 +44,9 @@ import java.io.{ByteArrayInputStream, ObjectOutputStream, ObjectStreamClass}
 import java.nio.charset.Charset
 import java.time._
 import java.util.UUID
-
 import scala.collection.{mutable => mut}
 import scala.collection.compat._
+import scala.collection.compat.immutable.ArraySeq
 import scala.collection.immutable.SortedMap
 import scala.jdk.CollectionConverters._
 
@@ -154,6 +155,20 @@ final class CoderTest extends AnyFlatSpec with Matchers {
       beOfType[CoderTransform[_, _]] and
       materializeTo[ArrayCoder[_]] and
       beFullyCompliantNotConsistentWithEquals()
+
+    val pqOrd: Ordering[String] = (x: String, y: String) => x.reverse.compareTo(y.reverse)
+    val pq = new mut.PriorityQueue[String]()(pqOrd)
+    pq ++= s
+
+    implicit val pqEq: Equality[mut.PriorityQueue[String]] = {
+      case (a: mut.PriorityQueue[String], b: mut.PriorityQueue[_]) => a.toList == b.toList
+      case _                                                       => false
+    }
+
+    pq coderShould roundtrip() and
+      beOfType[CoderTransform[_, _]] and
+      materializeTo[MutablePriorityQueueCoder[_]] and
+      beFullyCompliantNotConsistentWithEquals()
   }
 
   it should "support Scala enumerations" in {
@@ -321,7 +336,12 @@ final class CoderTest extends AnyFlatSpec with Matchers {
   }
 
   it should "support Java collections" in {
-    import java.util.{ArrayList => jArrayList, List => jList, Map => jMap}
+    import java.util.{
+      ArrayList => jArrayList,
+      List => jList,
+      Map => jMap,
+      PriorityQueue => jPriorityQueue
+    }
     val is = 1 to 10
     val s: jList[String] = is.map(_.toString).asJava
     val m: jMap[String, Int] = is
@@ -344,6 +364,23 @@ final class CoderTest extends AnyFlatSpec with Matchers {
       beOfType[Transform[_, _]] and
       materializeToTransformOf[beam.ListCoder[_]] and
       beFullyCompliant()
+
+    val pqOrd: Ordering[String] = (x: String, y: String) => x.reverse.compareTo(y.reverse)
+    val pq = new jPriorityQueue[String](pqOrd)
+    pq.addAll(s)
+    implicit val pqCoder: Coder[java.util.PriorityQueue[String]] =
+      Coder.jPriorityQueueCoder[String](pqOrd)
+
+    implicit val pqEq: Equality[java.util.PriorityQueue[String]] = {
+      case (a: jPriorityQueue[String], b: jPriorityQueue[_]) =>
+        ArraySeq.unsafeWrapArray(a.toArray) == ArraySeq.unsafeWrapArray(b.toArray)
+      case _ => false
+    }
+
+    pq coderShould roundtrip() and
+      beOfType[Transform[_, _]] and
+      materializeToTransformOf[ArrayCoder[_]] and
+      beFullyCompliantNotConsistentWithEquals()
   }
 
   it should "Derive serializable coders" in {