Include labeled post IDs in Under the Hood reports

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
+        )
     }
   }
 }

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)
     }
   }
 }

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)
+  }
+}

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(""))

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)
+    }
+  }
+}

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),
         }