scala 3 support for json streaming #146

pjfanning · 2023-06-08T22:15:56Z

Part of #126

A strange Scala 3 issue means that immutable.Queue.enqueue(ByteString) won't compile. I tried to produce a minimal case in scastie to report a Scala compiler bug but the minimal case works.

For now, I think using a mutable.Queue is ok and I can log an issue to keep checking when we upgrade Scala 3 to see if the issue is fixed.

nvollmar

lgtm

He-Pin · 2023-06-09T06:26:02Z

@pjfanning What's the error looks like.

mdedetrich · 2023-06-09T07:34:16Z

Yeah I am a bit wary of changing the underlying collection type from immutable to mutable. @pjfanning Can you post the error that you got?

pjfanning · 2023-06-09T07:51:17Z

[error] -- [E007] Type Mismatch Error: /Users/pj.fanning/code/incubator-pekko-connectors/json-streaming/src/main/scala/org/apache/pekko/stream/connectors/json/impl/JsonStreamReader.scala:52:48 
[error] 52 |              buffer = buffer.enqueue(ByteString(value.toString))
[error]    |                                      ^^^^^^^^^^^^^^^^^^^^^^^^^^
[error]    |                    Found:    org.apache.pekko.util.ByteString
[error]    |                    Required: Iterable[org.apache.pekko.util.ByteString]
[error]    |
[error]    | longer explanation available when compiling with `-explain`
[warn] three warnings found
[error] one error found
[error] (json-streaming / Compile / compileIncremental) Compilation failed

There is a deprecated enqueue(Iterable[T]) function and for some reason Scala 3.3.0 compiler ignores the fact enqueue(T) is available.

mdedetrich · 2023-06-09T07:54:00Z

There is a deprecated enqueue(Iterable[T]) function and for some reason Scala 3.3.0 compiler ignores the fact enqueue(T) is available.

What happens if you put ByteString(value.toString) into a List with a single value, i.e.

buffer.enqueue(List(ByteString(value.toString)))

?

pjfanning · 2023-06-09T07:56:36Z

There is a deprecated enqueue(Iterable[T]) function and for some reason Scala 3.3.0 compiler ignores the fact enqueue(T) is available.

What happens if you put ByteString(value.toString) into a List with a single value, i.e.

buffer.enqueue(List(ByteString(value.toString)))

?

That works but isn't it wasteful to create a Seq to get around the compiler issue?

mdedetrich

So I just checked the code and using a mutable queue to get around this problem is fine since its internal however there is one thing I noticed, can we change private var buffer = Queue.empty[ByteString] to a val (since there is no reason for it to be a var, we are not overriding the reference since we are using a mutable collection).

pjfanning · 2023-06-09T08:09:31Z

So I just checked the code and using a mutable queue to get around this problem is fine since its internal however there is one thing I noticed, can we change private var buffer = Queue.empty[ByteString] to a val (since there is no reason for it to be a var, we are not overriding the reference since we are using a mutable collection).

There is a 2nd place where the queue is recreated. I can see if I can clear the original queue instead of resetting the var.

mdedetrich · 2023-06-09T08:10:20Z

There is a 2nd place where the queue is recreated. I can see if I can clear the original queue instead of resetting the var.

You should be able to just clear the queue?

pjfanning · 2023-06-09T08:21:17Z

There is a 2nd place where the queue is recreated. I can see if I can clear the original queue instead of resetting the var.

You should be able to just clear the queue?

clearing a queue affects the iterator that was previously created for the queue (before the clear) - causes an IndexOutOfBoundsException
I'm trying a different solution at the moment

pjfanning · 2023-06-09T08:57:19Z

@mdedetrich could you review the latest change when you get a chance?

mdedetrich · 2023-06-09T09:13:49Z

...streaming/src/main/scala/org/apache/pekko/stream/connectors/json/impl/JsonStreamReader.scala

@@ -69,8 +69,8 @@ private[pekko] final class JsonStreamReader(path: JsonPath) extends GraphStage[F
        }

        if (buffer.nonEmpty) {
-          emitMultiple(out, buffer)
-          buffer = Queue.empty[ByteString]
+          emitMultiple(out, buffer.toSeq.iterator)


So I just noticed this now and I am thinking that the List(ByteString(value.toString)) variant is better because while there may be a trivial box for a single List (which might get inlined anyways), the buffer.toSeq.iterator is basically going to copy the entire collection which is much worse especially considering that the size of this collection can be quite large

@pjfanning You can revert this line now since buffer is immutable

mdedetrich · 2023-06-09T09:17:21Z

@pjfanning Ah I noticed the core problem, just use enqueueAll instead of enqueue (otherwise the solution is exactly the same)! ByteString is actually an iterator/collection and so enqueueAll is what we should be using, the definition of both of these functions should make it clear

  @deprecated("Use `enqueueAll` instead of `enqueue` to enqueue a collection of elements", "2.13.0")
  @`inline` final def enqueue[B >: A](iter: scala.collection.Iterable[B]) = enqueueAll(iter)

  /** Creates a new queue with all elements provided by an `Iterable` object
    *  added at the end of the old queue.
    *
    *  The elements are appended in the order they are given out by the
    *  iterator.
    *
    *  @param  iter        an iterable object
    */
  def enqueueAll[B >: A](iter: scala.collection.Iterable[B]): Queue[B] = appendedAll(iter)

In other words the current version is calling def enqueue[B >: A](iter: scala.collection.Iterable[B]) = enqueueAll(iter) which is deprecated in 2.13 and ByteString is a subtype of scala.collection.Iterable.

pjfanning · 2023-06-09T09:48:55Z

@pjfanning Ah I noticed the core problem, just use enqueueAll instead of enqueue (otherwise the solution is exactly the same)! ByteString is actually an iterator/collection and so enqueueAll is what we should be using, the definition of both of these functions should make it clear
  @deprecated("Use `enqueueAll` instead of `enqueue` to enqueue a collection of elements", "2.13.0")
  @`inline` final def enqueue[B >: A](iter: scala.collection.Iterable[B]) = enqueueAll(iter)

  /** Creates a new queue with all elements provided by an `Iterable` object
    *  added at the end of the old queue.
    *
    *  The elements are appended in the order they are given out by the
    *  iterator.
    *
    *  @param  iter        an iterable object
    */
  def enqueueAll[B >: A](iter: scala.collection.Iterable[B]): Queue[B] = appendedAll(iter)
In other words the current version is calling def enqueue[B >: A](iter: scala.collection.Iterable[B]) = enqueueAll(iter) which is deprecated in 2.13 and ByteString is a subtype of scala.collection.Iterable.

That doesn't compile in Scala 2.

[error] /Users/pj.fanning/code/incubator-pekko-connectors/json-streaming/src/main/scala/org/apache/pekko/stream/connectors/json/impl/JsonStreamReader.scala:52:52: type mismatch;
[error]  found   : org.apache.pekko.util.ByteString
[error]  required: Iterable[org.apache.pekko.util.ByteString]
[error]               buffer = buffer.enqueueAll(ByteString(value.toString))
[error]                                                    ^
[error] one error found

buffer is Queue[ByteString] so only ByteStrings can be queued. ByteString might be an Iterable[Byte] but that is not compatible with Queue[ByteString].

pjfanning · 2023-06-09T10:06:25Z

I raised scala/scala3#17946

...streaming/src/main/scala/org/apache/pekko/stream/connectors/json/impl/JsonStreamReader.scala

json-streaming/src/main/scala/org/apache/pekko/stream/connectors/json/impl/MyByteString.scala

Update JsonStreamReader.scala make buffer a val Update JsonStreamReader.scala Update JsonStreamReader.scala use Iterable.single Delete MyByteString.scala compile issue in scala 2.12 Update JsonStreamReader.scala

mdedetrich

Perfect, approving assuming CI passes.

mdedetrich · 2023-06-09T10:37:58Z

@pjfanning Just noticed that Iterable.single doesn't exist in Scala 2.12. I will fix this in Pekko.

pjfanning · 2023-06-09T10:50:44Z

@pjfanning Just noticed that Iterable.single doesn't exist in Scala 2.12. I will fix this in Pekko.

I'm trying to add a QueueHelper with different Scala 2 and Scala 3 implementations in the json-streaming module.

pjfanning · 2023-06-09T11:07:43Z

@pjfanning Just noticed that Iterable.single doesn't exist in Scala 2.12. I will fix this in Pekko.

I'm trying to add a QueueHelper with different Scala 2 and Scala 3 implementations in the json-streaming module.

@mdedetrich I added the QueueHelper

mdedetrich · 2023-06-09T11:15:29Z

I'm trying to add a QueueHelper with different Scala 2 and Scala 3 implementations in the json-streaming module.

I have already added the Iterable.single method to Scala 2.12 under pekko's ccompat module. The PR is here apache/pekko#378

mdedetrich · 2023-06-09T11:17:26Z

Okay so this solution is nicer in the sense that for Scala 2 its going to be faster/unchanged and only Scala 3 has the workaround. Still approved from me end.

pjfanning · 2023-06-09T11:18:40Z

Okay so this solution is nicer in the sense that for Scala 2 its going to be faster/unchanged and only Scala 3 has the workaround. Still approved from me end.

My solution has the benefit that in Scala 2, no wrapping Iterable is used

* scala 3 support for json streaming Update JsonStreamReader.scala make buffer a val Update JsonStreamReader.scala Update JsonStreamReader.scala use Iterable.single Delete MyByteString.scala compile issue in scala 2.12 Update JsonStreamReader.scala * Update Dependencies.scala * use separate scala2/3 code * Update QueueHelper.scala

pjfanning changed the base branch from main to scala3 June 8, 2023 22:46

pjfanning force-pushed the scala3 branch from 585cacb to c54ee93 Compare June 8, 2023 23:00

pjfanning requested review from He-Pin, nvollmar and mdedetrich June 8, 2023 23:27

nvollmar approved these changes Jun 9, 2023

View reviewed changes

mdedetrich requested changes Jun 9, 2023

View reviewed changes

mdedetrich reviewed Jun 9, 2023

View reviewed changes

pjfanning force-pushed the scala3-json-streaming branch from 3cc59be to e77ba9c Compare June 9, 2023 09:50

mdedetrich reviewed Jun 9, 2023

View reviewed changes

...streaming/src/main/scala/org/apache/pekko/stream/connectors/json/impl/JsonStreamReader.scala Outdated Show resolved Hide resolved

mdedetrich reviewed Jun 9, 2023

View reviewed changes

json-streaming/src/main/scala/org/apache/pekko/stream/connectors/json/impl/MyByteString.scala Outdated Show resolved Hide resolved

pjfanning force-pushed the scala3-json-streaming branch from ff88a98 to e44d75d Compare June 9, 2023 10:20

scala 3 support for json streaming

c2d6a41

Update JsonStreamReader.scala make buffer a val Update JsonStreamReader.scala Update JsonStreamReader.scala use Iterable.single Delete MyByteString.scala compile issue in scala 2.12 Update JsonStreamReader.scala

pjfanning force-pushed the scala3-json-streaming branch from 30d7c43 to c2d6a41 Compare June 9, 2023 10:26

Update Dependencies.scala

88a74cc

mdedetrich approved these changes Jun 9, 2023

View reviewed changes

pjfanning added 2 commits June 9, 2023 12:02

use separate scala2/3 code

28013d0

Update QueueHelper.scala

3c86077

mdedetrich mentioned this pull request Jun 9, 2023

Add Iterable.once to ccompat apache/pekko#378

Merged

pjfanning merged commit bb2dbdb into apache:scala3 Jun 9, 2023

pjfanning deleted the scala3-json-streaming branch June 9, 2023 11:48

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

scala 3 support for json streaming #146

scala 3 support for json streaming #146

pjfanning commented Jun 8, 2023 •

edited

Loading

nvollmar left a comment

He-Pin commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich left a comment •

edited

Loading

pjfanning commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich Jun 9, 2023

mdedetrich Jun 9, 2023

mdedetrich commented Jun 9, 2023 •

edited

Loading

pjfanning commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich left a comment

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

scala 3 support for json streaming #146

scala 3 support for json streaming #146

Conversation

pjfanning commented Jun 8, 2023 • edited Loading

nvollmar left a comment

Choose a reason for hiding this comment

He-Pin commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich left a comment • edited Loading

Choose a reason for hiding this comment

pjfanning commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich Jun 9, 2023

Choose a reason for hiding this comment

mdedetrich Jun 9, 2023

Choose a reason for hiding this comment

mdedetrich commented Jun 9, 2023 • edited Loading

pjfanning commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich left a comment

Choose a reason for hiding this comment

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

pjfanning commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

mdedetrich commented Jun 9, 2023

pjfanning commented Jun 9, 2023

pjfanning commented Jun 8, 2023 •

edited

Loading

mdedetrich left a comment •

edited

Loading

mdedetrich commented Jun 9, 2023 •

edited

Loading