diff --git a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala index 6739b139..fde521f0 100644 --- a/under-the-hood/scalding/UthDailyPostsBackfillJob.scala +++ b/under-the-hood/scalding/UthDailyPostsBackfillJob.scala @@ -100,7 +100,7 @@ class UthDailyPostsBackfillApp { observationMs, config) .map { - case (userId, authoredDay, label, asOfDay, carried, removed) => + case (userId, authoredDay, label, asOfDay, carried, removed, postIds) => val age = calendarDaysBetween(authoredDay, asOfDay) UthDailyPostLabel( userId = Some(userId), @@ -111,7 +111,8 @@ class UthDailyPostsBackfillApp { asOfYyyymmdd = Some(asOfDay), observationAgeDays = Some(age), isFinal = Some(age >= config.postObservationDays), - postObservationDays = Some(config.postObservationDays) + postObservationDays = Some(config.postObservationDays), + postIds = Some(postIds) ) } @@ -125,7 +126,7 @@ class UthDailyPostsBackfillApp { rangeEndMs, config.reducers) .map { - case (userId, authoredDay, label, carried) => + case (userId, authoredDay, label, carried, postIds) => UthDailyPostLabel( userId = Some(userId), authoredYyyymmdd = Some(authoredDay), @@ -135,7 +136,8 @@ class UthDailyPostsBackfillApp { asOfYyyymmdd = Some(authoredDay), observationAgeDays = Some(0), isFinal = Some(true), - postObservationDays = Some(config.postObservationDays) + postObservationDays = Some(config.postObservationDays), + postIds = Some(postIds) ) } @@ -240,7 +242,7 @@ object UthDailyPostsBackfillApp { rangeEndMs: Long, observationMs: Long, config: UthDailyPostsConfig - ): TypedPipe[(Long, Int, String, Int, Long, Long)] = { + ): TypedPipe[(Long, Int, String, Int, Long, Long, Seq[Long])] = { val postByTweetId = applyReducers( posts.map { case (tweetId, userId, day, logicalId, createdMs) => @@ -294,19 +296,29 @@ object UthDailyPostsBackfillApp { applyReducers( reduced.collect { - case ((_, userId, day, label, asOfDay, createdMs), (everApply, _, lastApply, lastExpires)) - if everApply => + case ( + (logicalId, userId, day, label, asOfDay, createdMs), + (everApply, _, lastApply, lastExpires) + ) if everApply => val deadline = math.min(createdMs + observationMs, yyyymmddToMs(asOfDay) + DayMs) val removed = if (UthDailyPostsApp .removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L else 0L - ((userId, day, label, asOfDay), (1L, removed)) + ((userId, day, label, asOfDay), UthPostIds.single(logicalId, removed)) }.group, config.reducers - ).sum.toTypedPipe.map { - case ((userId, day, label, asOfDay), (carried, removed)) => - (userId, day, label, asOfDay, carried, removed) + ).reduce(UthPostIds.merge).toTypedPipe.map { + case ((userId, day, label, asOfDay), summary) => + ( + userId, + day, + label, + asOfDay, + summary.carried, + summary.removed, + summary.postIds + ) } } } diff --git a/under-the-hood/scalding/UthDailyPostsJob.scala b/under-the-hood/scalding/UthDailyPostsJob.scala index dbafd530..35a3c9e6 100644 --- a/under-the-hood/scalding/UthDailyPostsJob.scala +++ b/under-the-hood/scalding/UthDailyPostsJob.scala @@ -107,7 +107,7 @@ class UthDailyPostsApp { observationMs, config) .map { - case (userId, authoredDay, label, carried, removed) => + case (userId, authoredDay, label, carried, removed, postIds) => val age = calendarDaysBetween(authoredDay, asOfDay) UthDailyPostLabel( userId = Some(userId), @@ -118,7 +118,8 @@ class UthDailyPostsApp { asOfYyyymmdd = Some(asOfDay), observationAgeDays = Some(age), isFinal = Some(age >= config.postObservationDays), - postObservationDays = Some(config.postObservationDays) + postObservationDays = Some(config.postObservationDays), + postIds = Some(postIds) ) } @@ -131,7 +132,7 @@ class UthDailyPostsApp { dayEndMs, config.reducers) .map { - case (userId, authoredDay, label, carried) => + case (userId, authoredDay, label, carried, postIds) => UthDailyPostLabel( userId = Some(userId), authoredYyyymmdd = Some(authoredDay), @@ -141,7 +142,8 @@ class UthDailyPostsApp { asOfYyyymmdd = Some(asOfDay), observationAgeDays = Some(0), isFinal = Some(true), - postObservationDays = Some(config.postObservationDays) + postObservationDays = Some(config.postObservationDays), + postIds = Some(postIds) ) } @@ -192,7 +194,7 @@ object UthDailyPostsApp { dayStartMs: Long, dayEndMs: Long, reducers: Int - ): TypedPipe[(Long, Int, String, Long)] = + ): TypedPipe[(Long, Int, String, Long, Seq[Long])] = if (flagLabels.isEmpty) TypedPipe.empty else { val flagged = tweets.flatMap { t => @@ -207,11 +209,17 @@ object UthDailyPostsApp { } else Nil } else Nil } - val counts = applyReducers(flagged.group, reducers).sum.keys.map { - case (userId, day, _, label) => ((userId, day, label), 1L) - }.sumByKey - (if (reducers > 0) counts.withReducers(reducers) else counts).toTypedPipe - .map { case ((userId, day, label), carried) => (userId, day, label, carried) } + val distinctPosts = applyReducers(flagged.group, reducers).sum.keys + applyReducers( + distinctPosts.map { + case (userId, day, logicalId, label) => + ((userId, day, label), UthPostIds.single(logicalId, 0L)) + }.group, + reducers + ).reduce(UthPostIds.merge).toTypedPipe.map { + case ((userId, day, label), summary) => + (userId, day, label, summary.carried, summary.postIds) + } } private[under_the_hood] def loadPosts( @@ -340,7 +348,7 @@ object UthDailyPostsApp { dayEndMs: Long, observationMs: Long, config: UthDailyPostsConfig - ): TypedPipe[(Long, Int, String, Long, Long)] = { + ): TypedPipe[(Long, Int, String, Long, Long, Seq[Long])] = { val postByTweetId = posts.map { case (tweetId, userId, day, logicalId, createdMs) => (tweetId, (logicalId, userId, day, createdMs)) @@ -380,17 +388,19 @@ object UthDailyPostsApp { applyReducers( reduced.collect { - case ((_, userId, day, label, createdMs), (everApply, _, lastApply, lastExpires)) - if everApply => + case ( + (logicalId, userId, day, label, createdMs), + (everApply, _, lastApply, lastExpires) + ) if everApply => val deadline = math.min(createdMs + observationMs, dayEndMs) val removed = if (removedAfterLastAction(lastApply, lastExpires, createdMs, deadline)) 1L else 0L - ((userId, day, label), (1L, removed)) + ((userId, day, label), UthPostIds.single(logicalId, removed)) }.group, config.reducers - ).sum.toTypedPipe.map { - case ((userId, day, label), (carried, removed)) => - (userId, day, label, carried, removed) + ).reduce(UthPostIds.merge).toTypedPipe.map { + case ((userId, day, label), summary) => + (userId, day, label, summary.carried, summary.removed, summary.postIds) } } } diff --git a/under-the-hood/scalding/UthPostIds.scala b/under-the-hood/scalding/UthPostIds.scala new file mode 100644 index 00000000..bbf104f8 --- /dev/null +++ b/under-the-hood/scalding/UthPostIds.scala @@ -0,0 +1,29 @@ +package com.twitter.visibility.under_the_hood + +object UthPostIds { + val MaxPostIdsPerLabel: Int = 1000 + + final case class Summary(carried: Long, removed: Long, postIds: Seq[Long]) + + val empty: Summary = Summary(0L, 0L, Vector.empty) + + def single(logicalId: Long, removed: Long): Summary = + Summary(1L, if (removed > 0L) 1L else 0L, Vector(logicalId)) + + def merge(a: Summary, b: Summary): Summary = + Summary( + carried = a.carried + b.carried, + removed = a.removed + b.removed, + postIds = newestDistinct(a.postIds ++ b.postIds, MaxPostIdsPerLabel) + ) + + def summarize(rows: Iterable[(Long, Long)]): Summary = + rows.foldLeft(empty) { + case (acc, (logicalId, removed)) => merge(acc, single(logicalId, removed)) + } + + def newestDistinct(postIds: Iterable[Long], limit: Int): Seq[Long] = { + require(limit > 0, s"limit must be > 0; got $limit") + postIds.toSet.toSeq.sorted.takeRight(limit) + } +} diff --git a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala index 31ed6037..c698b83d 100644 --- a/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala +++ b/under-the-hood/scalding/UthUserMonthMhPublisherJob.scala @@ -165,7 +165,11 @@ object UthUserMonthMhPublisherApp { label <- row.label carried <- row.carried removed <- row.removed - } yield ((userId, monthBucket(day)), (label, dayOfMonth(day), carried, removed)) + postIds = row.postIds.getOrElse(Seq.empty) + } yield ( + (userId, monthBucket(day)), + (label, dayOfMonth(day), carried, removed, postIds) + ) }, reducers ) @@ -202,21 +206,34 @@ object UthUserMonthMhPublisherApp { val postLabelAgg = labelsOpt .getOrElse(Nil) - .groupBy { case (label, _, _, _) => label } + .groupBy { case (label, _, _, _, _) => label } .map { case (label, rows) => - val days = rows - .groupBy { case (_, day, _, _) => day } - .map { - case (day, dayRows) => - val best = dayRows.maxBy { - case (_, _, carried, removed) => (carried, removed) - } - UthDayCarriedRemoved(Some(day), Some(best._3), Some(best._4)) + val selectedRows = rows + .groupBy { case (_, day, _, _, _) => day } + .values + .map { dayRows => + dayRows.maxBy { + case (_, _, carried, removed, postIds) => + (carried, removed, postIds.size) + } } .toList + val days = selectedRows + .map { + case (_, day, carried, removed, _) => + UthDayCarriedRemoved(Some(day), Some(carried), Some(removed)) + } .sortBy(_.dayOfMonth.getOrElse(0)) - UthPostLabelAggregate(Some(label), Some(days)) + val postIds = UthPostIds.newestDistinct( + selectedRows.flatMap(_._5), + UthPostIds.MaxPostIdsPerLabel + ) + UthPostLabelAggregate( + label = Some(label), + days = Some(days), + postIds = Some(postIds) + ) } .toList .sortBy(_.label.getOrElse("")) diff --git a/under-the-hood/src/test/scala/com/twitter/visibility/under_the_hood/UthPostIdsSpec.scala b/under-the-hood/src/test/scala/com/twitter/visibility/under_the_hood/UthPostIdsSpec.scala new file mode 100644 index 00000000..faf3eb9c --- /dev/null +++ b/under-the-hood/src/test/scala/com/twitter/visibility/under_the_hood/UthPostIdsSpec.scala @@ -0,0 +1,108 @@ +package com.twitter.visibility.under_the_hood + +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class UthPostIdsSpec extends AnyWordSpec with Matchers { + + private def summaryOf(rows: (Long, Long)*): UthPostIds.Summary = + UthPostIds.summarize(rows) + + "UthPostIds.single" should { + "count one carried post and normalize any positive removal to one" in { + UthPostIds.single(10L, 0L) shouldBe UthPostIds.Summary(1L, 0L, Vector(10L)) + UthPostIds.single(10L, 1L) shouldBe UthPostIds.Summary(1L, 1L, Vector(10L)) + UthPostIds.single(10L, 5L).removed shouldBe 1L + } + } + + "UthPostIds.merge" should { + "sum carried and removed the way the upstream summed rows" in { + val merged = UthPostIds.merge(UthPostIds.single(1L, 1L), UthPostIds.single(2L, 0L)) + merged.carried shouldBe 2L + merged.removed shouldBe 1L + merged.postIds shouldBe Seq(1L, 2L) + } + + "treat empty as an identity" in { + val one = UthPostIds.single(7L, 1L) + UthPostIds.merge(UthPostIds.empty, one) shouldBe one + UthPostIds.merge(one, UthPostIds.empty) shouldBe one + } + + "be associative and order independent so map-side combining is safe" in { + val a = UthPostIds.single(3L, 0L) + val b = UthPostIds.single(1L, 1L) + val c = UthPostIds.single(2L, 0L) + val left = UthPostIds.merge(UthPostIds.merge(a, b), c) + val right = UthPostIds.merge(a, UthPostIds.merge(b, c)) + left shouldBe right + UthPostIds.merge(a, b) shouldBe UthPostIds.merge(b, a) + } + + "never let a merged id list exceed the bound" in { + val big = UthPostIds.Summary( + carried = UthPostIds.MaxPostIdsPerLabel.toLong, + removed = 0L, + postIds = (1L to UthPostIds.MaxPostIdsPerLabel.toLong).toVector + ) + val other = UthPostIds.Summary(0L, 0L, (5000L to 5100L).toVector) + UthPostIds.merge(big, other).postIds.size shouldBe UthPostIds.MaxPostIdsPerLabel + } + } + + "UthPostIds.summarize" should { + "produce ids in deterministic ascending order" in { + summaryOf((30L, 0L), (10L, 0L), (20L, 0L)).postIds shouldBe Seq(10L, 20L, 30L) + } + + "reconcile removed counts against carried counts" in { + val s = summaryOf((1L, 1L), (2L, 0L), (3L, 1L)) + s.carried shouldBe 3L + s.removed shouldBe 2L + } + + "return the empty summary for no rows" in { + UthPostIds.summarize(Nil) shouldBe UthPostIds.Summary(0L, 0L, Vector.empty) + } + + "bound the id list while leaving carried counts exact" in { + val rows = (1L to 2500L).map(id => (id, 0L)) + val s = UthPostIds.summarize(rows) + s.carried shouldBe 2500L + s.postIds.size shouldBe UthPostIds.MaxPostIdsPerLabel + s.postIds.head shouldBe 1501L + s.postIds.last shouldBe 2500L + } + } + + "UthPostIds.newestDistinct" should { + "keep the newest ids by snowflake ordering" in { + UthPostIds.newestDistinct(Seq(5L, 1L, 9L, 3L), 2) shouldBe Seq(5L, 9L) + } + + "deduplicate before applying the bound" in { + UthPostIds.newestDistinct(Seq(4L, 4L, 4L, 1L), 10) shouldBe Seq(1L, 4L) + } + + "return everything when the bound is not reached" in { + UthPostIds.newestDistinct(Seq(2L, 1L), 10) shouldBe Seq(1L, 2L) + } + + "reject a non positive bound" in { + an[IllegalArgumentException] should be thrownBy UthPostIds.newestDistinct(Seq(1L), 0) + } + } + + "a per day cap" should { + "not change which ids survive the monthly cap" in { + val dayA = (1L to 2500L).map(id => (id, 0L)) + val dayB = (9000L to 9010L).map(id => (id, 0L)) + val cappedPerDay = + UthPostIds.summarize(dayA).postIds ++ UthPostIds.summarize(dayB).postIds + val uncapped = (dayA ++ dayB).map(_._1) + UthPostIds.newestDistinct(cappedPerDay, UthPostIds.MaxPostIdsPerLabel) shouldBe + UthPostIds.newestDistinct(uncapped, UthPostIds.MaxPostIdsPerLabel) + } + } +} diff --git a/under-the-hood/strato/columns/underTheHoodReport.User.strato b/under-the-hood/strato/columns/underTheHoodReport.User.strato index 6be87004..87de5a46 100644 --- a/under-the-hood/strato/columns/underTheHoodReport.User.strato +++ b/under-the-hood/strato/columns/underTheHoodReport.User.strato @@ -190,11 +190,14 @@ def buildReportJson( .flatMap { agg => agg.label.filter(underTheHoodLabels.isPostLabel).map { raw => val posts = sumCarried(agg.days) + val postIds = agg.postIds.getOrElse(Seq.empty).distinct.sorted { label = underTheHoodLabels.postLabelName(raw), about = underTheHoodLabels.postLabelAbout(raw), effect = underTheHoodLabels.postLabelEffect(raw), posts = posts, + postIds = postIds.map(_.toString), + postIdsComplete = postIds.size.toLong == posts, totalPostsInMonth = postCount, percentageOfPosts = formatPercentage(posts, postCount), } diff --git a/under-the-hood/thrift/uth_serving.thrift b/under-the-hood/thrift/uth_serving.thrift index 299c551f..5fa52f91 100644 --- a/under-the-hood/thrift/uth_serving.thrift +++ b/under-the-hood/thrift/uth_serving.thrift @@ -53,6 +53,8 @@ enum UthFollowerClass { struct UthPostLabelAggregate { 1: optional string label (personalDataType = 'TweetSafetyLabels') 2: optional list days + // Newest logical post IDs represented by days, bounded by the month publisher. + 4: optional list postIds (personalDataType = 'TweetId') }(persisted = 'true', hasPersonalData = 'true') struct UthAccountLabelAggregate { @@ -129,6 +131,8 @@ struct UthDailyPostLabel { 7: optional i32 observationAgeDays 8: optional bool isFinal 9: optional i32 postObservationDays + // Logical post IDs represented by carried; absent on pre-migration rows. + 11: optional list postIds (personalDataType = 'TweetId') }(persisted = 'true', hasPersonalData = 'true') struct UthDailyAccountLabel {