Features for 0.2.0 #29
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
istreeter
force-pushed
the
pubsub-sink-wrap-bytes
branch
4 times, most recently
from
72d55e4 to
b522173
Compare
istreeter
force-pushed
the
pubsub-sink-wrap-bytes
branch
from
b522173 to
bb2367c
Compare
istreeter
changed the title
benjben
approved these changes
Nov 24, 2023
| */ | ||
| package com.google.protobuf | ||
|
|
||
| object UnsafeSnowplowOps { |
modules/streams-core/src/main/scala/com.snowplowanalytics.snowplow/sinks/ListOfList.scala
Outdated
Show resolved
Hide resolved
modules/streams-core/src/main/scala/com.snowplowanalytics.snowplow/sinks/ListOfList.scala
Outdated
Show resolved
Hide resolved
| * The resulting `ListOfList` does not have the same order as the input List. This is helpful in | ||
| * Snowplow apps where order of events within batches is not important. | ||
| */ | ||
| def mapUnordered[B](f: A => B): ListOfList[B] = |
There was a problem hiding this comment.
Suggested change
| def mapUnordered[B](f: A => B): ListOfList[B] = | |
| def map[B](f: A => B): ListOfList[B] = |
Even though we don't care, the order ends up being preserved
There was a problem hiding this comment.
It is not preserved!
scala> val lol = ListOfList.of(List(List(1,2,3), List(4,5,6)))
val lol: com.snowplowanalytics.snowplow.sinks.ListOfList[Int] = com.snowplowanalytics.snowplow.sinks.ListOfList@3fdd4a80
scala> lol.copyToIndexedSeq
val res1: IndexedSeq[Int] = Vector(1, 2, 3, 4, 5, 6)
scala> lol.mapUnordered(identity).copyToIndexedSeq
val res2: IndexedSeq[Int] = Vector(6, 5, 4, 3, 2, 1)
There was a problem hiding this comment.
I get it now, it is because of .asIterable.toList that it is preserved here:
def nonEmpty3 = {
val input = ListOfList.of(List(List(1, 2, 3), List(4, 5, 6)))
val expected = List(100, 200, 300, 400, 500, 600)
val result = input.mapUnordered(_ * 100).asIterable.toList
result must containTheSameElementsAs(expected)
}
There was a problem hiding this comment.
Oh I see the confusion, but that's just a specs2 thing.
In specs2, containTheSameElementsAs only tests that all elements are present in any order. The documentation is here and it contains this example:
Finally, if you want to get the differences between 2 traversables:
Seq(2, 4, 1) must containTheSameElementsAs(Seq(1, 4, 2))
Comment on lines
113
to
128
| private case class PendingBatch[B](value: Option[B], weight: Long) | ||
|
|
||
| private object PendingBatch { | ||
| def empty[B]: PendingBatch[B] = PendingBatch(None, 0L) | ||
| } | ||
|
|
||
| /** | ||
| * The result of combining a chunk of `A`s, while not exceeding total weight. | ||
| * | ||
| * @param notAtSize | ||
| * Optionally an `A` that does not yet exceed the maximum allowed size. We should not emit this | ||
| * `A` but instead wait in case we can combine it with other `A`s later. | ||
| * Optionally a batch `B` that does not yet exceed the maximum allowed size. We should not emit | ||
| * this `B` but instead wait in case we can combine it with other `A`s later. | ||
| * @param toEmit | ||
| * The combined `A`s which meet size requirements. These should be emitted downstream because we | ||
| * The combined `B`s which meet size requirements. These should be emitted downstream because we | ||
| * cannot combine them with anything more. | ||
| */ | ||
| private case class CombineByWeightResult[A](notAtSize: Option[A], toEmit: Vector[A]) | ||
| private case class CombineByWeightResult[B]( | ||
| doNotEmitYet: Option[B], | ||
| pendingWeight: Long, | ||
| toEmit: Vector[B] | ||
| ) |
There was a problem hiding this comment.
I find it a bit hard to follow to have both a PendingBatch and a CombineByWeightResult which is supposed to be The result of combining a chunk but that can nevertheless be incomplete
There was a problem hiding this comment.
Good point. I re-worked the classes and return types a bit, which I think makes it neater now.
| private case class PendingBatch[B](value: Option[B], weight: Long) | ||
|
|
||
| private object PendingBatch { | ||
| def empty[B]: PendingBatch[B] = PendingBatch(None, 0L) | ||
| } | ||
|
|
||
| /** | ||
| * The result of combining a chunk of `A`s, while not exceeding total weight. | ||
| * | ||
| * @param notAtSize |
There was a problem hiding this comment.
Suggested change
| * @param notAtSize | |
| * @param doNotEmitYet |
istreeter
force-pushed
the
pubsub-sink-wrap-bytes
branch
from
bb2367c to
b326dd4
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
No description provided.