Add domainRepoId to Subdomain class (#674)

* Change Subdomain class to contain domainRepoId

* Remove jpaTm from Spec11PipelineTest and change clientId -> registrarId

* Remove 'client' from a comment

* Include changes to Spec11Pipeline

* add SafeBrowsingTransforms

* Run style

core/src/main/java/google/registry/beam/spec11/SafeBrowsingTransforms.java

@@ -141,7 +141,7 @@ public class SafeBrowsingTransforms {
     @ProcessElement
     public void processElement(ProcessContext context) {
       Subdomain subdomain = context.element();
-      subdomainBuffer.put(subdomain.fullyQualifiedDomainName(), subdomain);
+      subdomainBuffer.put(subdomain.domainName(), subdomain);
       if (subdomainBuffer.size() >= BATCH_SIZE) {
         ImmutableSet<KV<Subdomain, ThreatMatch>> results = evaluateAndFlush();
         results.forEach(context::output);
@@ -239,7 +239,7 @@ public class SafeBrowsingTransforms {
             String url = match.getJSONObject("threat").getString("url");
             Subdomain subdomain = subdomainBuffer.get(url);
             resultBuilder.add(
-                KV.of(subdomain, ThreatMatch.create(match, subdomain.fullyQualifiedDomainName())));
+                KV.of(subdomain, ThreatMatch.create(match, subdomain.domainName())));
           }
         }
       }

core/src/main/java/google/registry/beam/spec11/Spec11Pipeline.java

@@ -77,7 +77,7 @@ public class Spec11Pipeline implements Serializable {
   public static final String REGISTRAR_EMAIL_FIELD = "registrarEmailAddress";
   /** The JSON object field into which we put the registrar's name for Spec11 reports. */
   public static final String REGISTRAR_CLIENT_ID_FIELD = "registrarClientId";
-  /** The JSON object field we put the threat match array for Spec11 reports. */
+  /** The JSON object field into which we put the threat match array for Spec11 reports. */
   public static final String THREAT_MATCHES_FIELD = "threatMatches";
 
   private final String projectId;
@@ -176,9 +176,11 @@ public class Spec11Pipeline implements Serializable {
       PCollection<Subdomain> domains,
       EvaluateSafeBrowsingFn evaluateSafeBrowsingFn,
       ValueProvider<String> dateProvider) {
-    PCollection<KV<Subdomain, ThreatMatch>> subdomains =
+
+    /* Store ThreatMatch objects in JSON. */
+    PCollection<KV<Subdomain, ThreatMatch>> subdomainsJson =
         domains.apply("Run through SafeBrowsingAPI", ParDo.of(evaluateSafeBrowsingFn));
-    subdomains
+    subdomainsJson
         .apply(
             "Map registrar client ID to email/ThreatMatch pair",
             MapElements.into(
@@ -187,7 +189,7 @@ public class Spec11Pipeline implements Serializable {
                 .via(
                     (KV<Subdomain, ThreatMatch> kv) ->
                         KV.of(
-                            kv.getKey().registrarClientId(),
+                            kv.getKey().registrarId(),
                             EmailAndThreatMatch.create(
                                 kv.getKey().registrarEmailAddress(), kv.getValue()))))
         .apply("Group by registrar client ID", GroupByKey.create())

core/src/main/java/google/registry/beam/spec11/Subdomain.java

@@ -36,12 +36,14 @@ import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
 public abstract class Subdomain implements Serializable {
 
   private static final ImmutableList<String> FIELD_NAMES =
-      ImmutableList.of("fullyQualifiedDomainName", "registrarClientId", "registrarEmailAddress");
+      ImmutableList.of("domainName", "domainRepoId", "registrarId", "registrarEmailAddress");
 
   /** Returns the fully qualified domain name. */
-  abstract String fullyQualifiedDomainName();
-  /** Returns the client ID of the associated registrar for this domain. */
-  abstract String registrarClientId();
+  abstract String domainName();
+  /** Returns the domain repo ID (the primary key of the domain table). */
+  abstract String domainRepoId();
+  /** Returns the registrar ID of the associated registrar for this domain. */
+  abstract String registrarId();
   /** Returns the email address of the registrar associated with this domain. */
   abstract String registrarEmailAddress();
 
@@ -56,8 +58,9 @@ public abstract class Subdomain implements Serializable {
     checkFieldsNotNull(FIELD_NAMES, schemaAndRecord);
     GenericRecord record = schemaAndRecord.getRecord();
     return create(
-        extractField(record, "fullyQualifiedDomainName"),
-        extractField(record, "registrarClientId"),
+        extractField(record, "domainName"),
+        extractField(record, "domainRepoId"),
+        extractField(record, "registrarId"),
         extractField(record, "registrarEmailAddress"));
   }
 
@@ -69,9 +72,11 @@ public abstract class Subdomain implements Serializable {
    */
   @VisibleForTesting
   static Subdomain create(
-      String fullyQualifiedDomainName, String registrarClientId, String registrarEmailAddress) {
+      String domainName,
+      String domainRepoId,
+      String registrarId,
+      String registrarEmailAddress) {
     return new AutoValue_Subdomain(
-        fullyQualifiedDomainName, registrarClientId, registrarEmailAddress);
+        domainName, domainRepoId, registrarId, registrarEmailAddress);
   }
 }
-

core/src/main/java/google/registry/beam/spec11/sql/subdomains.sql

@@ -19,11 +19,13 @@
   -- email address.
 
 SELECT
-  domain.fullyQualifiedDomainName AS fullyQualifiedDomainName,
-  registrar.clientId AS registrarClientId,
+  domain.fullyQualifiedDomainName AS domainName,
+  domain.__key__.name AS domainRepoId,
+  registrar.clientId AS clientId,
   COALESCE(registrar.emailAddress, '') AS registrarEmailAddress
 FROM ( (
     SELECT
+      __key__,
       fullyQualifiedDomainName,
       currentSponsorClientId,
       creationTime