From 50fae3302527fc2566b18fac6b655bffa2e69485 Mon Sep 17 00:00:00 2001 From: jnadeau207-collab Date: Wed, 2 Sep 2026 16:13:36 -0400 Subject: [PATCH] under-the-hood: include labeled post ids in reports Carry logical post IDs through the daily and backfill post-label rows, persist them in optional Thrift fields, aggregate the newest 1,000 distinct IDs per label into monthly rows, and emit them as strings in reportJson alongside postIdsComplete. IDs use the same logical-post identity (initialTweetId.getOrElse(tweetId)) as the existing carried counts, so an edited post chain stays one post in both the count and the ID list. Aggregation keeps the existing map-side combining: UthPostIds.merge is an associative, commutative reduce that sums carried and removed exactly as the previous sum did, and bounds the ID list at every merge. Peak reducer memory per key stays bounded rather than growing with posts per key, and the daily persisted rows carry the same bound as the monthly rows. Field IDs 4 and 11 are intentional, leaving 3 and 10 for the source-provenance fields proposed in #52. --- .../scalding/UthDailyPostsBackfillJob.scala | 34 ++++-- .../scalding/UthDailyPostsJob.scala | 44 ++++--- under-the-hood/scalding/UthPostIds.scala | 29 +++++ .../scalding/UthUserMonthMhPublisherJob.scala | 39 +++++-- .../under_the_hood/UthPostIdsSpec.scala | 108 ++++++++++++++++++ .../columns/underTheHoodReport.User.strato | 3 + under-the-hood/thrift/uth_serving.thrift | 4 + 7 files changed, 222 insertions(+), 39 deletions(-) create mode 100644 under-the-hood/scalding/UthPostIds.scala create mode 100644 under-the-hood/src/test/scala/com/twitter/visibility/under_the_hood/UthPostIdsSpec.scala 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 {