Regression in Scala 3 for propagation of @uncheckedVariance via dependent types

[mdedetrich]

Compiler version

Scala 3.3.0

Minimized code

SplitAfterSize(chunkSize, chunkBufferSize)(atLeastOneByteString)
  .via(getChunkBuffer(chunkSize, chunkBufferSize, maxRetries)) // creates the chunks
  .mergeSubstreamsWithParallelism(parallelism)
  .filter(_.size > 0)
  .via(atLeastOne)
  .zip(requestInfoOrUploadState(s3Location, contentType, s3Headers, initialUploadState))

Output

[error] -- Error: /Users/mdedetrich/github/incubator-pekko-connectors/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/S3Stream.scala:1184:11 
[error] 1180 |        SplitAfterSize(chunkSize, chunkBufferSize)(atLeastOneByteString)
[error] 1181 |          .via(getChunkBuffer(chunkSize, chunkBufferSize, maxRetries)) // creates the chunks
[error] 1182 |          .mergeSubstreamsWithParallelism(parallelism)
[error] 1183 |          .filter(_.size > 0)
[error] 1184 |          .via(atLeastOne)
[error]      |        ^
[error]      |bad parameter reference ([O] =>>
[error]      |  org.apache.pekko.stream.scaladsl.Flow[
[error]      |    org.apache.pekko.util.ByteString @uncheckedVariance, O,
[error]      |    org.apache.pekko.NotUsed @uncheckedVariance]
[error]      |) @uncheckedVariance[org.apache.pekko.stream.connectors.s3.impl.Chunk]#Mat at sbt-api
[error]      |the parameter is type Mat in class Flow but the prefix ([O] =>>
[error]      |  org.apache.pekko.stream.scaladsl.Flow[
[error]      |    org.apache.pekko.util.ByteString @uncheckedVariance, O,
[error]      |    org.apache.pekko.NotUsed @uncheckedVariance]
[error]      |) @uncheckedVariance[org.apache.pekko.stream.connectors.s3.impl.Chunk]
[error]      |does not define any corresponding arguments.
[error]      |idx = 2, args = org.apache.pekko.stream.connectors.s3.impl.Chunk,
[error]      |constraint =  uninstantiated variables:
[error]      | constrained types:
[error]      | bounds:
[error]      | ordering:
[error]      | co-deps:
[error]      | contra-deps:

Expectation

That the above code compiles without any errors just as it does in Scala 2.

Note that it is possible to avoid this issue by adding explicit type annotations, i.e. the following variant compiles fine

val source1: SubFlow[Chunk, NotUsed, Flow[ByteString, ByteString, NotUsed]#Repr, Sink[ByteString, NotUsed]] =
  SplitAfterSize(chunkSize, chunkBufferSize)(atLeastOneByteString)
    .via(getChunkBuffer(chunkSize, chunkBufferSize, maxRetries)) // creates the chunks

val source2 = source1.mergeSubstreamsWithParallelism(parallelism)
  .filter(_.size > 0)
  .via(atLeastOne)

source2
  .zip(requestInfoOrUploadState(s3Location, contentType, s3Headers, initialUploadState))

I have a hunch that the cause of this compiler error is due to @uncheckedVariance at https://github.com/apache/incubator-pekko/blob/580f12c29fb61b65758af9b0c31494af0af17175/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala#L69 not being properly propagated.

I haven't had time to properly minimize the example since as you can see it occured in a fairly complex usecase but I will try and get around to it if I have time.

[Gedochao]

@mdedetrich did you find the time to minimise this?
And if not, can you confirm if this is still valid on 3.3.1 and/or 3.4.0-RC4?

[mdedetrich]

@Gedochao It does indeed seem to be solved for Scala 3.3.1, see apache/pekko-connectors#464 . Feel free to close the issue.

[mdedetrich]

Ill just close issue myself as solved in Scala 3.3.1