Add Cloud SQL queries for transaction reports (#1397)

* Add the Cloud SQL queries for transaction reports

* Add the remaining queries

* Some query fixes

* Fix comments

* Fix indentation in total_nameservers

* Fix indentation on other Case condition

core/src/main/java/google/registry/reporting/icann/TransactionsReportingQueryBuilder.java

@@ -14,6 +14,7 @@
 
 package google.registry.reporting.icann;
 
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.reporting.icann.IcannReportingModule.DATASTORE_EXPORT_DATA_SET;
 import static google.registry.reporting.icann.IcannReportingModule.ICANN_REPORTING_DATA_SET;
 import static google.registry.reporting.icann.QueryBuilderUtils.getQueryFromFile;
@@ -71,102 +72,180 @@ public final class TransactionsReportingQueryBuilder implements QueryBuilder {
     DateTime latestReportTime = earliestReportTime.plusMonths(1).minusMillis(1);
 
     ImmutableMap.Builder<String, String> queriesBuilder = ImmutableMap.builder();
-    String registrarIanaIdQuery =
+    String registrarIanaIdQuery;
-        SqlTemplate.create(getQueryFromFile("registrar_iana_id.sql"))
+    if (tm().isOfy()) {
-            .put("PROJECT_ID", projectId)
+      registrarIanaIdQuery =
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+          SqlTemplate.create(getQueryFromFile("registrar_iana_id.sql"))
-            .put("REGISTRAR_TABLE", "Registrar")
+              .put("PROJECT_ID", projectId)
-            .build();
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+              .put("REGISTRAR_TABLE", "Registrar")
+              .build();
+    } else {
+      registrarIanaIdQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_registrar_iana_id.sql"))
+              .put("PROJECT_ID", projectId)
+              .build();
+    }
     queriesBuilder.put(getTableName(REGISTRAR_IANA_ID, yearMonth), registrarIanaIdQuery);
 
-    String totalDomainsQuery =
+    String totalDomainsQuery;
-        SqlTemplate.create(getQueryFromFile("total_domains.sql"))
+    if (tm().isOfy()) {
-            .put("PROJECT_ID", projectId)
+      totalDomainsQuery =
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+          SqlTemplate.create(getQueryFromFile("total_domains.sql"))
-            .put("DOMAINBASE_TABLE", "DomainBase")
+              .put("PROJECT_ID", projectId)
-            .put("REGISTRAR_TABLE", "Registrar")
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .build();
+              .put("DOMAINBASE_TABLE", "DomainBase")
+              .put("REGISTRAR_TABLE", "Registrar")
+              .build();
+    } else {
+      totalDomainsQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_total_domains.sql"))
+              .put("PROJECT_ID", projectId)
+              .build();
+    }
     queriesBuilder.put(getTableName(TOTAL_DOMAINS, yearMonth), totalDomainsQuery);
 
     DateTimeFormatter timestampFormatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS");
-    String totalNameserversQuery =
+    String totalNameserversQuery;
-        SqlTemplate.create(getQueryFromFile("total_nameservers.sql"))
+    if (tm().isOfy()) {
-            .put("PROJECT_ID", projectId)
+      totalNameserversQuery =
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+          SqlTemplate.create(getQueryFromFile("total_nameservers.sql"))
-            .put("HOSTRESOURCE_TABLE", "HostResource")
+              .put("PROJECT_ID", projectId)
-            .put("DOMAINBASE_TABLE", "DomainBase")
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .put("REGISTRAR_TABLE", "Registrar")
+              .put("HOSTRESOURCE_TABLE", "HostResource")
-            .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
+              .put("DOMAINBASE_TABLE", "DomainBase")
-            .build();
+              .put("REGISTRAR_TABLE", "Registrar")
+              .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
+              .build();
+    } else {
+      totalNameserversQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_total_nameservers.sql"))
+              .put("PROJECT_ID", projectId)
+              .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
+              .build();
+    }
     queriesBuilder.put(getTableName(TOTAL_NAMESERVERS, yearMonth), totalNameserversQuery);
 
-    String transactionCountsQuery =
+    String transactionCountsQuery;
-        SqlTemplate.create(getQueryFromFile("transaction_counts.sql"))
+    if (tm().isOfy()) {
-            .put("PROJECT_ID", projectId)
+      transactionCountsQuery =
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+          SqlTemplate.create(getQueryFromFile("transaction_counts.sql"))
-            .put("REGISTRAR_TABLE", "Registrar")
+              .put("PROJECT_ID", projectId)
-            .put("HISTORYENTRY_TABLE", "HistoryEntry")
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .put("EARLIEST_REPORT_TIME", timestampFormatter.print(earliestReportTime))
+              .put("REGISTRAR_TABLE", "Registrar")
-            .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
+              .put("HISTORYENTRY_TABLE", "HistoryEntry")
-            .put("CLIENT_ID", "clientId")
+              .put("EARLIEST_REPORT_TIME", timestampFormatter.print(earliestReportTime))
-            .put("OTHER_CLIENT_ID", "otherClientId")
+              .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
-            .put("TRANSFER_SUCCESS_FIELD", "TRANSFER_GAINING_SUCCESSFUL")
+              .put("CLIENT_ID", "clientId")
-            .put("TRANSFER_NACKED_FIELD", "TRANSFER_GAINING_NACKED")
+              .put("OTHER_CLIENT_ID", "otherClientId")
-            .put("DEFAULT_FIELD", "field")
+              .put("TRANSFER_SUCCESS_FIELD", "TRANSFER_GAINING_SUCCESSFUL")
-            .build();
+              .put("TRANSFER_NACKED_FIELD", "TRANSFER_GAINING_NACKED")
+              .put("DEFAULT_FIELD", "field")
+              .build();
+    } else {
+      transactionCountsQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_transaction_counts.sql"))
+              .put("PROJECT_ID", projectId)
+              .put("EARLIEST_REPORT_TIME", timestampFormatter.print(earliestReportTime))
+              .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
+              .build();
+    }
     queriesBuilder.put(getTableName(TRANSACTION_COUNTS, yearMonth), transactionCountsQuery);
 
-    String transactionTransferLosingQuery =
+    String transactionTransferLosingQuery;
-        SqlTemplate.create(getQueryFromFile("transaction_counts.sql"))
+    if (tm().isOfy()) {
-            .put("PROJECT_ID", projectId)
+      transactionTransferLosingQuery =
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+          SqlTemplate.create(getQueryFromFile("transaction_counts.sql"))
-            .put("REGISTRAR_TABLE", "Registrar")
+              .put("PROJECT_ID", projectId)
-            .put("HISTORYENTRY_TABLE", "HistoryEntry")
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .put("EARLIEST_REPORT_TIME", timestampFormatter.print(earliestReportTime))
+              .put("REGISTRAR_TABLE", "Registrar")
-            .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
+              .put("HISTORYENTRY_TABLE", "HistoryEntry")
-            // Roles are reversed for losing queries
+              .put("EARLIEST_REPORT_TIME", timestampFormatter.print(earliestReportTime))
-            .put("CLIENT_ID", "otherClientId")
+              .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
-            .put("OTHER_CLIENT_ID", "clientId")
+              // Roles are reversed for losing queries
-            .put("TRANSFER_SUCCESS_FIELD", "TRANSFER_LOSING_SUCCESSFUL")
+              .put("CLIENT_ID", "otherClientId")
-            .put("TRANSFER_NACKED_FIELD", "TRANSFER_LOSING_NACKED")
+              .put("OTHER_CLIENT_ID", "clientId")
-            .put("DEFAULT_FIELD", "NULL")
+              .put("TRANSFER_SUCCESS_FIELD", "TRANSFER_LOSING_SUCCESSFUL")
-            .build();
+              .put("TRANSFER_NACKED_FIELD", "TRANSFER_LOSING_NACKED")
+              .put("DEFAULT_FIELD", "NULL")
+              .build();
+    } else {
+      transactionTransferLosingQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_transaction_transfer_losing.sql"))
+              .put("PROJECT_ID", projectId)
+              .put("EARLIEST_REPORT_TIME", timestampFormatter.print(earliestReportTime))
+              .put("LATEST_REPORT_TIME", timestampFormatter.print(latestReportTime))
+              .build();
+    }
     queriesBuilder.put(
         getTableName(TRANSACTION_TRANSFER_LOSING, yearMonth), transactionTransferLosingQuery);
 
     // App Engine log table suffixes use YYYYMMDD format
     DateTimeFormatter logTableFormatter = DateTimeFormat.forPattern("yyyyMMdd");
-    String attemptedAddsQuery =
+    String attemptedAddsQuery;
-        SqlTemplate.create(getQueryFromFile("attempted_adds.sql"))
+    if (tm().isOfy()) {
-            .put("PROJECT_ID", projectId)
+      attemptedAddsQuery =
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+          SqlTemplate.create(getQueryFromFile("attempted_adds.sql"))
-            .put("REGISTRAR_TABLE", "Registrar")
+              .put("PROJECT_ID", projectId)
-            .put("APPENGINE_LOGS_DATA_SET", "appengine_logs")
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .put("REQUEST_TABLE", "appengine_googleapis_com_request_log_")
+              .put("REGISTRAR_TABLE", "Registrar")
-            .put("FIRST_DAY_OF_MONTH", logTableFormatter.print(earliestReportTime))
+              .put("APPENGINE_LOGS_DATA_SET", "appengine_logs")
-            .put("LAST_DAY_OF_MONTH", logTableFormatter.print(latestReportTime))
+              .put("REQUEST_TABLE", "appengine_googleapis_com_request_log_")
-            // All metadata logs for reporting come from google.registry.flows.FlowReporter.
+              .put("FIRST_DAY_OF_MONTH", logTableFormatter.print(earliestReportTime))
-            .put(
+              .put("LAST_DAY_OF_MONTH", logTableFormatter.print(latestReportTime))
-                "METADATA_LOG_PREFIX",
+              // All metadata logs for reporting come from google.registry.flows.FlowReporter.
-                "google.registry.flows.FlowReporter recordToLogs: FLOW-LOG-SIGNATURE-METADATA")
+              .put(
-            .build();
+                  "METADATA_LOG_PREFIX",
+                  "google.registry.flows.FlowReporter recordToLogs: FLOW-LOG-SIGNATURE-METADATA")
+              .build();
+    } else {
+      attemptedAddsQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_attempted_adds.sql"))
+              .put("PROJECT_ID", projectId)
+              .put("APPENGINE_LOGS_DATA_SET", "appengine_logs")
+              .put("REQUEST_TABLE", "appengine_googleapis_com_request_log_")
+              .put("FIRST_DAY_OF_MONTH", logTableFormatter.print(earliestReportTime))
+              .put("LAST_DAY_OF_MONTH", logTableFormatter.print(latestReportTime))
+              // All metadata logs for reporting come from google.registry.flows.FlowReporter.
+              .put(
+                  "METADATA_LOG_PREFIX",
+                  "google.registry.flows.FlowReporter recordToLogs: FLOW-LOG-SIGNATURE-METADATA")
+              .build();
+    }
     queriesBuilder.put(getTableName(ATTEMPTED_ADDS, yearMonth), attemptedAddsQuery);
 
-    String aggregateQuery =
+    String aggregateQuery;
-        SqlTemplate.create(getQueryFromFile("transactions_report_aggregation.sql"))
+    if (tm().isOfy()) {
-            .put("PROJECT_ID", projectId)
+      aggregateQuery =
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+          SqlTemplate.create(getQueryFromFile("transactions_report_aggregation.sql"))
-            .put("REGISTRY_TABLE", "Registry")
+              .put("PROJECT_ID", projectId)
-            .put("ICANN_REPORTING_DATA_SET", icannReportingDataSet)
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .put("REGISTRAR_IANA_ID_TABLE", getTableName(REGISTRAR_IANA_ID, yearMonth))
+              .put("REGISTRY_TABLE", "Registry")
-            .put("TOTAL_DOMAINS_TABLE", getTableName(TOTAL_DOMAINS, yearMonth))
+              .put("ICANN_REPORTING_DATA_SET", icannReportingDataSet)
-            .put("TOTAL_NAMESERVERS_TABLE", getTableName(TOTAL_NAMESERVERS, yearMonth))
+              .put("REGISTRAR_IANA_ID_TABLE", getTableName(REGISTRAR_IANA_ID, yearMonth))
-            .put("TRANSACTION_COUNTS_TABLE", getTableName(TRANSACTION_COUNTS, yearMonth))
+              .put("TOTAL_DOMAINS_TABLE", getTableName(TOTAL_DOMAINS, yearMonth))
-            .put(
+              .put("TOTAL_NAMESERVERS_TABLE", getTableName(TOTAL_NAMESERVERS, yearMonth))
-                "TRANSACTION_TRANSFER_LOSING_TABLE",
+              .put("TRANSACTION_COUNTS_TABLE", getTableName(TRANSACTION_COUNTS, yearMonth))
-                getTableName(TRANSACTION_TRANSFER_LOSING, yearMonth))
+              .put(
-            .put("ATTEMPTED_ADDS_TABLE", getTableName(ATTEMPTED_ADDS, yearMonth))
+                  "TRANSACTION_TRANSFER_LOSING_TABLE",
-            .build();
+                  getTableName(TRANSACTION_TRANSFER_LOSING, yearMonth))
+              .put("ATTEMPTED_ADDS_TABLE", getTableName(ATTEMPTED_ADDS, yearMonth))
+              .build();
+    } else {
+      aggregateQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_transactions_report_aggregation.sql"))
+              .put("PROJECT_ID", projectId)
+              .put("ICANN_REPORTING_DATA_SET", icannReportingDataSet)
+              .put("REGISTRAR_IANA_ID_TABLE", getTableName(REGISTRAR_IANA_ID, yearMonth))
+              .put("TOTAL_DOMAINS_TABLE", getTableName(TOTAL_DOMAINS, yearMonth))
+              .put("TOTAL_NAMESERVERS_TABLE", getTableName(TOTAL_NAMESERVERS, yearMonth))
+              .put("TRANSACTION_COUNTS_TABLE", getTableName(TRANSACTION_COUNTS, yearMonth))
+              .put(
+                  "TRANSACTION_TRANSFER_LOSING_TABLE",
+                  getTableName(TRANSACTION_TRANSFER_LOSING, yearMonth))
+              .put("ATTEMPTED_ADDS_TABLE", getTableName(ATTEMPTED_ADDS, yearMonth))
+              .build();
+    }
     queriesBuilder.put(getTableName(TRANSACTIONS_REPORT_AGGREGATION, yearMonth), aggregateQuery);
 
     return queriesBuilder.build();

core/src/main/java/google/registry/reporting/icann/sql/cloud_sql_attempted_adds.sql

@@ -0,0 +1,73 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Determine the number of attempted adds each registrar made.
+
+  -- Since the specification requests all 'attempted' adds, we regex the
+  -- monthly App Engine logs, searching for all create commands and associating
+  -- them with their corresponding registrars.
+
+  -- Example log generated by FlowReporter in App Engine logs:
+  --google.registry.flows.FlowReporter
+  -- recordToLogs: FLOW-LOG-SIGNATURE-METADATA:
+  --{"serverTrid":"oNwL2J2eRya7bh7c9oHIzg==-2360a","clientId":"ipmirror"
+  -- ,"commandType":"hello", "resourceType":"","flowClassName":"HelloFlow"
+  -- ,"targetId":"","targetIds":[],"tld":"",
+  -- "tlds":[],"icannActivityReportField":""}
+
+-- This outer select just converts the registrar's clientId to their name.
+SELECT
+  tld,
+  registrar_table.registrar_name AS registrar_name,
+  'ATTEMPTED_ADDS' AS metricName,
+  count AS metricValue
+FROM (
+  SELECT
+    JSON_EXTRACT_SCALAR(json, '$.tld') AS tld,
+    JSON_EXTRACT_SCALAR(json, '$.clientId') AS clientId,
+    COUNT(json) AS count
+  FROM (
+      -- Extract JSON metadata package from monthly logs
+    SELECT
+      REGEXP_EXTRACT(logMessages, r'FLOW-LOG-SIGNATURE-METADATA: (.*)\n?$')
+          AS json
+    FROM (
+      SELECT
+        protoPayload.resource AS requestPath,
+        ARRAY(
+          SELECT logMessage
+          FROM UNNEST(protoPayload.line)) AS logMessage
+      FROM
+        `%PROJECT_ID%.%APPENGINE_LOGS_DATA_SET%.%REQUEST_TABLE%*`
+      WHERE _TABLE_SUFFIX
+      BETWEEN '%FIRST_DAY_OF_MONTH%'
+      AND '%LAST_DAY_OF_MONTH%')
+    JOIN UNNEST(logMessage) AS logMessages
+    -- Look for metadata logs from epp and registrar console requests
+    WHERE requestPath IN ('/_dr/epp', '/_dr/epptool', '/registrar-xhr')
+    AND STARTS_WITH(logMessages, "%METADATA_LOG_PREFIX%")
+    -- Look for domain creates
+    AND REGEXP_CONTAINS(
+       logMessages, r'"commandType":"create","resourceType":"domain"')
+    -- Filter prober data
+    AND NOT REGEXP_CONTAINS(
+        logMessages, r'"prober-[a-z]{2}-((any)|(canary))"') )
+  GROUP BY tld, clientId ) AS logs_table
+JOIN
+  EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+              '''SELECT registrar_id, registrar_name FROM "Registrar";''') AS registrar_table
+  ON
+      logs_table.clientId = registrar_table.registrar_id
+  ORDER BY tld, registrar_name

core/src/main/java/google/registry/reporting/icann/sql/cloud_sql_registrar_iana_id.sql

@@ -0,0 +1,37 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Gather a list of all tld-registrar pairs, with their IANA IDs.
+
+  -- This establishes which registrars will appear in the reports.
+
+SELECT
+  allowedTlds AS tld,
+  registrar_name,
+  iana_identifier AS iana_id
+FROM
+  EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+  '''SELECT
+    allowedTlds,
+    registrar_name,
+    iana_identifier
+   FROM
+    "Registrar" AS r,
+    UNNEST(allowed_tlds) AS allowedTlds
+   WHERE
+    r.type='REAL' OR r.type='INTERNAL';''')
+-- Filter out prober data
+WHERE NOT ENDS_WITH(allowedTlds, "test")
+ORDER BY tld, registrar_name

core/src/main/java/google/registry/reporting/icann/sql/cloud_sql_total_domains.sql

@@ -0,0 +1,43 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Determine the number of domains each registrar sponsors per tld.
+
+  -- This is just the number of fullyQualifiedDomainNames under each
+  -- tld-registrar pair.
+
+SELECT
+  tld,
+  registrar_name,
+  'TOTAL_DOMAINS' as metricName,
+  COUNT(domain_name) as metricValue
+FROM
+  EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+  '''SELECT
+    registrar_name,
+    registrar_id
+  FROM "Registrar" AS r
+  WHERE r.type='REAL' OR r.type='INTERNAL';''')
+JOIN
+  EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+  '''SELECT
+    tld,
+    current_sponsor_registrar_id,
+    domain_name
+  FROM "Domain" AS d;''')
+ON
+  current_sponsor_registrar_id = registrar_id
+GROUP BY tld, registrar_name
+ORDER BY tld, registrar_name

core/src/main/java/google/registry/reporting/icann/sql/cloud_sql_total_nameservers.sql

@@ -0,0 +1,80 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Determine the number of referenced nameservers for a registrar's domains.
+
+  -- We count the number of unique hosts under each tld-registrar combo by
+  -- collecting all domains' listed hosts that were still valid at the
+  -- end of the reporting month.
+
+SELECT
+    tld,
+    registrar_name,
+    'TOTAL_NAMESERVERS' AS metricName,
+    COUNT(host_name) AS metricValue
+FROM
+    EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+    '''SELECT
+        host_name,
+        current_sponsor_registrar_id,
+        creation_time AS host_creation_time,
+        CASE
+          WHEN deletion_time > to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+            THEN to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+          ELSE
+            deletion_time
+          END as host_deletion_time
+    FROM "Host";''')
+JOIN (
+    SELECT
+    registrar_id,
+    registrar_name
+    FROM
+        EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql", '''SELECT registrar_id, registrar_name FROM "Registrar" AS r WHERE r.type='REAL' OR r.type='INTERNAL';'''))
+ON current_sponsor_registrar_id = registrar_id
+JOIN (
+    SELECT
+        tld,
+        referencedHostName
+    FROM
+        EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+        '''SELECT
+                repo_id,
+                tld,
+                creation_time AS domain_creation_time,
+                CASE
+                  WHEN deletion_time > to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+                    THEN to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+                  ELSE
+                    deletion_time
+                  END as domain_deletion_time
+            FROM "Domain";''')
+        JOIN
+        EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql", '''SELECT domain_repo_id, host_repo_id FROM "DomainHost";''')
+        ON
+        repo_id = domain_repo_id
+        JOIN
+        EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql", '''SELECT repo_id as host_table_repo_id, host_name AS referencedHostName FROM "Host";''')
+        ON
+        host_table_repo_id = host_repo_id
+        WHERE
+        domain_creation_time <= TIMESTAMP("%LATEST_REPORT_TIME%")
+        AND domain_deletion_time > TIMESTAMP("%LATEST_REPORT_TIME%"))
+    ON host_name = referencedHostName
+WHERE
+    host_creation_time <= TIMESTAMP("%LATEST_REPORT_TIME%")
+    AND host_deletion_time > TIMESTAMP("%LATEST_REPORT_TIME%")
+GROUP BY tld, registrar_name
+ORDER BY tld, registrar_name

core/src/main/java/google/registry/reporting/icann/sql/cloud_sql_transaction_counts.sql

@@ -0,0 +1,72 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Counts the number of mutating transactions each registrar made.
+
+  -- We populate the fields through explicit logging of
+  -- DomainTransactionRecords, which contain all necessary information for
+  -- reporting (such as reporting time, report field, report amount, etc.
+
+SELECT
+    tld,
+    registrar_table.registrar_name AS registrar_name,
+    metricName,
+    metricValue
+FROM (
+    SELECT
+        tld,
+        clientId,
+        CASE WHEN field = 'TRANSFER_SUCCESSFUL' THEN 'TRANSFER_GAINING_SUCCESSFUL'
+        WHEN field = 'TRANSFER_NACKED' THEN 'TRANSFER_GAINING_NACKED'
+        ELSE field
+        END AS metricName,
+        SUM(amount) AS metricValue
+    FROM (
+        SELECT
+            CASE
+                -- Explicit transfer acks (approve) and nacks (reject) are done
+                -- by the opposing registrar. Thus, for these specific actions,
+                -- we swap the 'history_other_registrar_id' with the
+                -- 'history_registrar_id' to properly account for this reversal.
+                WHEN (history_type = 'DOMAIN_TRANSFER_APPROVE'
+                OR history_type = 'DOMAIN_TRANSFER_REJECT')
+                THEN history_other_registrar_id
+                ELSE history_registrar_id
+            END AS clientId,
+            tld,
+            report_field AS field,
+            report_amount AS amount,
+            reporting_time AS reportingTime
+            FROM EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+            ''' SELECT history_type, history_other_registrar_id, history_registrar_id, domain_repo_id, history_revision_id FROM "DomainHistory";''') AS dh
+            JOIN
+                EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+                '''SELECT domain_repo_id, history_revision_id, reporting_time, tld, report_field, report_amount FROM "DomainTransactionRecord";''') AS dtr
+            ON
+            dh.domain_repo_id = dtr.domain_repo_id AND dh.history_revision_id = dtr.history_revision_id
+        )
+    -- Only look at this month's data
+    WHERE reportingTime
+    BETWEEN TIMESTAMP('%EARLIEST_REPORT_TIME%')
+    AND TIMESTAMP('%LATEST_REPORT_TIME%')
+    GROUP BY
+        tld,
+        clientId,
+        field ) AS counts_table
+JOIN
+     EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+                '''SELECT registrar_id, registrar_name FROM "Registrar";''') AS registrar_table
+ON
+    counts_table.clientId = registrar_table.registrar_id

core/src/main/java/google/registry/reporting/icann/sql/cloud_sql_transaction_transfer_losing.sql

@@ -0,0 +1,72 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Counts the number of mutating transactions each registrar made.
+
+  -- We populate the fields through explicit logging of
+  -- DomainTransactionRecords, which contain all necessary information for
+  -- reporting (such as reporting time, report field, report amount, etc.
+
+SELECT
+    tld,
+    registrar_table.registrar_name AS registrar_name,
+    metricName,
+    metricValue
+FROM (
+    SELECT
+        tld,
+        clientId,
+        CASE WHEN field = 'TRANSFER_SUCCESSFUL' THEN 'TRANSFER_LOSING_SUCCESSFUL'
+        WHEN field = 'TRANSFER_NACKED' THEN 'TRANSFER_LOSING_NACKED'
+        ELSE NULL
+        END AS metricName,
+        SUM(amount) AS metricValue
+    FROM (
+        SELECT
+            CASE
+                -- Explicit transfer acks (approve) and nacks (reject) are done
+                -- by the opposing registrar. Thus, for these specific actions,
+                -- we swap the 'history_other_registrar_id' with the
+                -- 'history_registrar_id' to properly account for this reversal.
+                WHEN (history_type = 'DOMAIN_TRANSFER_APPROVE'
+                OR history_type = 'DOMAIN_TRANSFER_REJECT')
+                THEN history_registrar_id
+                ELSE history_other_registrar_id
+            END AS clientId,
+            tld,
+            report_field AS field,
+            report_amount AS amount,
+            reporting_time AS reportingTime
+            FROM EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+            ''' SELECT history_type, history_other_registrar_id, history_registrar_id, domain_repo_id, history_revision_id FROM "DomainHistory";''') AS dh
+            JOIN
+                EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+                '''SELECT domain_repo_id, history_revision_id, reporting_time, tld, report_field, report_amount FROM "DomainTransactionRecord";''') AS dtr
+            ON
+            dh.domain_repo_id = dtr.domain_repo_id AND dh.history_revision_id = dtr.history_revision_id
+        )
+    -- Only look at this month's data
+    WHERE reportingTime
+    BETWEEN TIMESTAMP('%EARLIEST_REPORT_TIME%')
+    AND TIMESTAMP('%LATEST_REPORT_TIME%')
+    GROUP BY
+        tld,
+        clientId,
+        field ) AS counts_table
+JOIN
+     EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql",
+                '''SELECT registrar_id, registrar_name FROM "Registrar";''') AS registrar_table
+ON
+    counts_table.clientId = registrar_table.registrar_id

core/src/main/java/google/registry/reporting/icann/sql/cloud_sql_transactions_report_aggregation.sql

@@ -0,0 +1,97 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Construct the transaction reports' rows from the intermediary data views.
+
+  -- This query pulls from all intermediary tables to create the activity
+  -- report csv, via a table transpose and sum over all activity report fields.
+
+SELECT
+  tlds.tld_name as tld,
+  -- Surround registrar names with quotes to handle names containing a comma.
+  FORMAT("\"%s\"", registrars.registrar_name) as registrar_name,
+  registrars.iana_id as iana_id,
+  SUM(IF(metrics.metricName = 'TOTAL_DOMAINS', metrics.metricValue, 0)) AS total_domains,
+  SUM(IF(metrics.metricName = 'TOTAL_NAMESERVERS', metrics.metricValue, 0)) AS total_nameservers,
+  SUM(IF(metrics.metricName = 'NET_ADDS_1_YR', metrics.metricValue, 0)) AS net_adds_1_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_2_YR', metrics.metricValue, 0)) AS net_adds_2_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_3_YR', metrics.metricValue, 0)) AS net_adds_3_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_4_YR', metrics.metricValue, 0)) AS net_adds_4_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_5_YR', metrics.metricValue, 0)) AS net_adds_5_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_6_YR', metrics.metricValue, 0)) AS net_adds_6_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_7_YR', metrics.metricValue, 0)) AS net_adds_7_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_8_YR', metrics.metricValue, 0)) AS net_adds_8_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_9_YR', metrics.metricValue, 0)) AS net_adds_9_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_10_Yr', metrics.metricValue, 0)) AS net_adds_10_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_1_YR', metrics.metricValue, 0)) AS net_renews_1_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_2_YR', metrics.metricValue, 0)) AS net_renews_2_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_3_YR', metrics.metricValue, 0)) AS net_renews_3_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_4_YR', metrics.metricValue, 0)) AS net_renews_4_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_5_YR', metrics.metricValue, 0)) AS net_renews_5_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_6_YR', metrics.metricValue, 0)) AS net_renews_6_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_7_YR', metrics.metricValue, 0)) AS net_renews_7_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_8_YR', metrics.metricValue, 0)) AS net_renews_8_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_9_YR', metrics.metricValue, 0)) AS net_renews_9_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_10_YR', metrics.metricValue, 0)) AS net_renews_10_yr,
+  SUM(IF(metrics.metricName = 'TRANSFER_GAINING_SUCCESSFUL', metrics.metricValue, 0)) AS transfer_gaining_successful,
+  SUM(IF(metrics.metricName = 'TRANSFER_GAINING_NACKED', metrics.metricValue, 0)) AS transfer_gaining_nacked,
+  SUM(IF(metrics.metricName = 'TRANSFER_LOSING_SUCCESSFUL', metrics.metricValue, 0)) AS transfer_losing_successful,
+  SUM(IF(metrics.metricName = 'TRANSFER_LOSING_NACKED', metrics.metricValue, 0)) AS transfer_losing_nacked,
+  -- We don't interact with transfer disputes
+  0 AS transfer_disputed_won,
+  0 AS transfer_disputed_lost,
+  0 AS transfer_disputed_nodecision,
+  SUM(IF(metrics.metricName = 'DELETED_DOMAINS_GRACE', metrics.metricValue, 0)) AS deleted_domains_grace,
+  SUM(IF(metrics.metricName = 'DELETED_DOMAINS_NOGRACE', metrics.metricValue, 0)) AS deleted_domains_nograce,
+  SUM(IF(metrics.metricName = 'RESTORED_DOMAINS', metrics.metricValue, 0)) AS restored_domains,
+  -- We don't require restore reports
+  0 AS restored_noreport,
+  -- We don't enforce AGP limits right now
+  0 AS agp_exemption_requests,
+  0 AS agp_exemptions_granted,
+  0 AS agp_exempted_domains,
+  SUM(IF(metrics.metricName = 'ATTEMPTED_ADDS', metrics.metricValue, 0)) AS attempted_adds
+FROM
+ -- Only produce reports for real TLDs
+    EXTERNAL_QUERY("projects/%PROJECT_ID%/locations/us/connections/%PROJECT_ID%-sql", '''SELECT "tld_name" FROM "Tld" AS t WHERE t.tld_type='REAL';''') AS tlds
+JOIN
+(SELECT *
+  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%REGISTRAR_IANA_ID_TABLE%`)
+  AS registrars
+ON tlds.tld_name = registrars.tld
+-- We LEFT JOIN to produce reports even if the registrar made no transactions
+LEFT OUTER JOIN (
+  -- Gather all intermediary data views
+  SELECT *
+  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%TOTAL_DOMAINS_TABLE%`
+  UNION ALL
+  SELECT *
+  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%TOTAL_NAMESERVERS_TABLE%`
+  UNION ALL
+  SELECT *
+  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%TRANSACTION_COUNTS_TABLE%`
+  UNION ALL
+  SELECT *
+  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%TRANSACTION_TRANSFER_LOSING_TABLE%`
+  UNION ALL
+  SELECT *
+  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%ATTEMPTED_ADDS_TABLE%` ) AS metrics
+-- Join on tld and registrar name
+ON registrars.tld = metrics.tld
+AND registrars.registrar_name = metrics.registrar_name
+GROUP BY
+tld, registrar_name, iana_id
+ORDER BY
+tld, registrar_name

core/src/test/java/google/registry/reporting/icann/TransactionsReportingQueryBuilderTest.java

@@ -18,31 +18,54 @@ import static com.google.common.truth.Truth.assertThat;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import google.registry.testing.AppEngineExtension;
+import google.registry.testing.DualDatabaseTest;
+import google.registry.testing.TestOfyOnly;
+import google.registry.testing.TestSqlOnly;
 import org.joda.time.YearMonth;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
 
 /** Unit tests for {@link ActivityReportingQueryBuilder}. */
+@DualDatabaseTest
 class TransactionsReportingQueryBuilderTest {
 
+  @RegisterExtension
+  public final AppEngineExtension appEngine =
+      AppEngineExtension.builder()
+          .withDatastoreAndCloudSql()
+          .withLocalModules()
+          .withTaskQueue()
+          .build();
+
   private final YearMonth yearMonth = new YearMonth(2017, 9);
 
-  private TransactionsReportingQueryBuilder getQueryBuilder() {
+  private TransactionsReportingQueryBuilder createQueryBuilder(String datasetName) {
-    TransactionsReportingQueryBuilder queryBuilder =
+    return new TransactionsReportingQueryBuilder("domain-registry-alpha", datasetName);
-        new TransactionsReportingQueryBuilder("domain-registry-alpha", "icann_reporting");
-    return queryBuilder;
   }
 
-  @Test
+  @TestOfyOnly
-  void testAggregateQueryMatch() {
+  void testAggregateQueryMatch_datastore() {
-    TransactionsReportingQueryBuilder queryBuilder = getQueryBuilder();
+    TransactionsReportingQueryBuilder queryBuilder = createQueryBuilder("icann_reporting");
     assertThat(queryBuilder.getReportQuery(yearMonth))
         .isEqualTo(
             "#standardSQL\nSELECT * FROM "
                 + "`domain-registry-alpha.icann_reporting.transactions_report_aggregation_201709`");
   }
 
-  @Test
+  @TestSqlOnly
-  void testIntermediaryQueryMatch() {
+  void testAggregateQueryMatch_cloud_sql() {
+    TransactionsReportingQueryBuilder queryBuilder =
+        createQueryBuilder("cloud_sql_icann_reporting");
+    assertThat(queryBuilder.getReportQuery(yearMonth))
+        .isEqualTo(
+            "#standardSQL\n"
+                + "SELECT * FROM "
+                + "`domain-registry-alpha.cloud_sql_icann_reporting"
+                + ".transactions_report_aggregation_201709`");
+  }
+
+  @TestOfyOnly
+  void testIntermediaryQueryMatch_datastore() {
     ImmutableList<String> expectedQueryNames =
         ImmutableList.of(
             TransactionsReportingQueryBuilder.TRANSACTIONS_REPORT_AGGREGATION,
@@ -53,7 +76,7 @@ class TransactionsReportingQueryBuilderTest {
             TransactionsReportingQueryBuilder.TRANSACTION_TRANSFER_LOSING,
             TransactionsReportingQueryBuilder.ATTEMPTED_ADDS);
 
-    TransactionsReportingQueryBuilder queryBuilder = getQueryBuilder();
+    TransactionsReportingQueryBuilder queryBuilder = createQueryBuilder("icann_reporting");
     ImmutableMap<String, String> actualQueries = queryBuilder.getViewQueryMap(yearMonth);
     for (String queryName : expectedQueryNames) {
       String actualTableName = String.format("%s_201709", queryName);
@@ -62,6 +85,30 @@ class TransactionsReportingQueryBuilderTest {
           .isEqualTo(ReportingTestData.loadFile(testFilename));
     }
   }
+
+  @TestSqlOnly
+  void testIntermediaryQueryMatch_cloud_sql() {
+    ImmutableList<String> expectedQueryNames =
+        ImmutableList.of(
+            TransactionsReportingQueryBuilder.REGISTRAR_IANA_ID,
+            TransactionsReportingQueryBuilder.TOTAL_DOMAINS,
+            TransactionsReportingQueryBuilder.TOTAL_NAMESERVERS,
+            TransactionsReportingQueryBuilder.TRANSACTION_COUNTS,
+            TransactionsReportingQueryBuilder.TRANSACTION_TRANSFER_LOSING,
+            TransactionsReportingQueryBuilder.ATTEMPTED_ADDS,
+            TransactionsReportingQueryBuilder.TRANSACTIONS_REPORT_AGGREGATION);
+
+    TransactionsReportingQueryBuilder queryBuilder =
+        createQueryBuilder("cloud_sql_icann_reporting");
+    ;
+    ImmutableMap<String, String> actualQueries = queryBuilder.getViewQueryMap(yearMonth);
+    for (String queryName : expectedQueryNames) {
+      String actualTableName = String.format("%s_201709", queryName);
+      String testFilename = String.format("%s_test_cloud_sql.sql", queryName);
+      assertThat(actualQueries.get(actualTableName))
+          .isEqualTo(ReportingTestData.loadFile(testFilename));
+    }
+  }
 }
 
 

core/src/test/resources/google/registry/reporting/icann/attempted_adds_test_cloud_sql.sql

@@ -0,0 +1,73 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Determine the number of attempted adds each registrar made.
+
+  -- Since the specification requests all 'attempted' adds, we regex the
+  -- monthly App Engine logs, searching for all create commands and associating
+  -- them with their corresponding registrars.
+
+  -- Example log generated by FlowReporter in App Engine logs:
+  --google.registry.flows.FlowReporter
+  -- recordToLogs: FLOW-LOG-SIGNATURE-METADATA:
+  --{"serverTrid":"oNwL2J2eRya7bh7c9oHIzg==-2360a","clientId":"ipmirror"
+  -- ,"commandType":"hello", "resourceType":"","flowClassName":"HelloFlow"
+  -- ,"targetId":"","targetIds":[],"tld":"",
+  -- "tlds":[],"icannActivityReportField":""}
+
+-- This outer select just converts the registrar's clientId to their name.
+SELECT
+  tld,
+  registrar_table.registrar_name AS registrar_name,
+  'ATTEMPTED_ADDS' AS metricName,
+  count AS metricValue
+FROM (
+  SELECT
+    JSON_EXTRACT_SCALAR(json, '$.tld') AS tld,
+    JSON_EXTRACT_SCALAR(json, '$.clientId') AS clientId,
+    COUNT(json) AS count
+  FROM (
+      -- Extract JSON metadata package from monthly logs
+    SELECT
+      REGEXP_EXTRACT(logMessages, r'FLOW-LOG-SIGNATURE-METADATA: (.*)\n?$')
+          AS json
+    FROM (
+      SELECT
+        protoPayload.resource AS requestPath,
+        ARRAY(
+          SELECT logMessage
+          FROM UNNEST(protoPayload.line)) AS logMessage
+      FROM
+        `domain-registry-alpha.appengine_logs.appengine_googleapis_com_request_log_*`
+      WHERE _TABLE_SUFFIX
+      BETWEEN '20170901'
+      AND '20170930')
+    JOIN UNNEST(logMessage) AS logMessages
+    -- Look for metadata logs from epp and registrar console requests
+    WHERE requestPath IN ('/_dr/epp', '/_dr/epptool', '/registrar-xhr')
+    AND STARTS_WITH(logMessages, "google.registry.flows.FlowReporter recordToLogs: FLOW-LOG-SIGNATURE-METADATA")
+    -- Look for domain creates
+    AND REGEXP_CONTAINS(
+       logMessages, r'"commandType":"create","resourceType":"domain"')
+    -- Filter prober data
+    AND NOT REGEXP_CONTAINS(
+        logMessages, r'"prober-[a-z]{2}-((any)|(canary))"') )
+  GROUP BY tld, clientId ) AS logs_table
+JOIN
+  EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+              '''SELECT registrar_id, registrar_name FROM "Registrar";''') AS registrar_table
+  ON
+      logs_table.clientId = registrar_table.registrar_id
+  ORDER BY tld, registrar_name

core/src/test/resources/google/registry/reporting/icann/registrar_iana_id_test_cloud_sql.sql

@@ -0,0 +1,37 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Gather a list of all tld-registrar pairs, with their IANA IDs.
+
+  -- This establishes which registrars will appear in the reports.
+
+SELECT
+  allowedTlds AS tld,
+  registrar_name,
+  iana_identifier AS iana_id
+FROM
+  EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+  '''SELECT
+    allowedTlds,
+    registrar_name,
+    iana_identifier
+   FROM
+    "Registrar" AS r,
+    UNNEST(allowed_tlds) AS allowedTlds
+   WHERE
+    r.type='REAL' OR r.type='INTERNAL';''')
+-- Filter out prober data
+WHERE NOT ENDS_WITH(allowedTlds, "test")
+ORDER BY tld, registrar_name

core/src/test/resources/google/registry/reporting/icann/total_domains_test_cloud_sql.sql

@@ -0,0 +1,43 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Determine the number of domains each registrar sponsors per tld.
+
+  -- This is just the number of fullyQualifiedDomainNames under each
+  -- tld-registrar pair.
+
+SELECT
+  tld,
+  registrar_name,
+  'TOTAL_DOMAINS' as metricName,
+  COUNT(domain_name) as metricValue
+FROM
+  EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+  '''SELECT
+    registrar_name,
+    registrar_id
+  FROM "Registrar" AS r
+  WHERE r.type='REAL' OR r.type='INTERNAL';''')
+JOIN
+  EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+  '''SELECT
+    tld,
+    current_sponsor_registrar_id,
+    domain_name
+  FROM "Domain" AS d;''')
+ON
+  current_sponsor_registrar_id = registrar_id
+GROUP BY tld, registrar_name
+ORDER BY tld, registrar_name

core/src/test/resources/google/registry/reporting/icann/total_nameservers_test_cloud_sql.sql

@@ -0,0 +1,80 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Determine the number of referenced nameservers for a registrar's domains.
+
+  -- We count the number of unique hosts under each tld-registrar combo by
+  -- collecting all domains' listed hosts that were still valid at the
+  -- end of the reporting month.
+
+SELECT
+    tld,
+    registrar_name,
+    'TOTAL_NAMESERVERS' AS metricName,
+    COUNT(host_name) AS metricValue
+FROM
+    EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+    '''SELECT
+        host_name,
+        current_sponsor_registrar_id,
+        creation_time AS host_creation_time,
+        CASE
+          WHEN deletion_time > to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+            THEN to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+          ELSE
+            deletion_time
+          END as host_deletion_time
+    FROM "Host";''')
+JOIN (
+    SELECT
+    registrar_id,
+    registrar_name
+    FROM
+        EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql", '''SELECT registrar_id, registrar_name FROM "Registrar" AS r WHERE r.type='REAL' OR r.type='INTERNAL';'''))
+ON current_sponsor_registrar_id = registrar_id
+JOIN (
+    SELECT
+        tld,
+        referencedHostName
+    FROM
+        EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+        '''SELECT
+                repo_id,
+                tld,
+                creation_time AS domain_creation_time,
+                CASE
+                  WHEN deletion_time > to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+                    THEN to_timestamp('9999-12-31 23:59:59', 'YYYY-MM-DD HH24:MI:SS')
+                  ELSE
+                    deletion_time
+                  END as domain_deletion_time
+            FROM "Domain";''')
+        JOIN
+        EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql", '''SELECT domain_repo_id, host_repo_id FROM "DomainHost";''')
+        ON
+        repo_id = domain_repo_id
+        JOIN
+        EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql", '''SELECT repo_id as host_table_repo_id, host_name AS referencedHostName FROM "Host";''')
+        ON
+        host_table_repo_id = host_repo_id
+        WHERE
+        domain_creation_time <= TIMESTAMP("2017-09-30 23:59:59.999")
+        AND domain_deletion_time > TIMESTAMP("2017-09-30 23:59:59.999"))
+    ON host_name = referencedHostName
+WHERE
+    host_creation_time <= TIMESTAMP("2017-09-30 23:59:59.999")
+    AND host_deletion_time > TIMESTAMP("2017-09-30 23:59:59.999")
+GROUP BY tld, registrar_name
+ORDER BY tld, registrar_name

core/src/test/resources/google/registry/reporting/icann/transaction_counts_test_cloud_sql.sql

@@ -0,0 +1,72 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Counts the number of mutating transactions each registrar made.
+
+  -- We populate the fields through explicit logging of
+  -- DomainTransactionRecords, which contain all necessary information for
+  -- reporting (such as reporting time, report field, report amount, etc.
+
+SELECT
+    tld,
+    registrar_table.registrar_name AS registrar_name,
+    metricName,
+    metricValue
+FROM (
+    SELECT
+        tld,
+        clientId,
+        CASE WHEN field = 'TRANSFER_SUCCESSFUL' THEN 'TRANSFER_GAINING_SUCCESSFUL'
+        WHEN field = 'TRANSFER_NACKED' THEN 'TRANSFER_GAINING_NACKED'
+        ELSE field
+        END AS metricName,
+        SUM(amount) AS metricValue
+    FROM (
+        SELECT
+            CASE
+                -- Explicit transfer acks (approve) and nacks (reject) are done
+                -- by the opposing registrar. Thus, for these specific actions,
+                -- we swap the 'history_other_registrar_id' with the
+                -- 'history_registrar_id' to properly account for this reversal.
+                WHEN (history_type = 'DOMAIN_TRANSFER_APPROVE'
+                OR history_type = 'DOMAIN_TRANSFER_REJECT')
+                THEN history_other_registrar_id
+                ELSE history_registrar_id
+            END AS clientId,
+            tld,
+            report_field AS field,
+            report_amount AS amount,
+            reporting_time AS reportingTime
+            FROM EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+            ''' SELECT history_type, history_other_registrar_id, history_registrar_id, domain_repo_id, history_revision_id FROM "DomainHistory";''') AS dh
+            JOIN
+                EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+                '''SELECT domain_repo_id, history_revision_id, reporting_time, tld, report_field, report_amount FROM "DomainTransactionRecord";''') AS dtr
+            ON
+            dh.domain_repo_id = dtr.domain_repo_id AND dh.history_revision_id = dtr.history_revision_id
+        )
+    -- Only look at this month's data
+    WHERE reportingTime
+    BETWEEN TIMESTAMP('2017-09-01 00:00:00.000')
+    AND TIMESTAMP('2017-09-30 23:59:59.999')
+    GROUP BY
+        tld,
+        clientId,
+        field ) AS counts_table
+JOIN
+     EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+                '''SELECT registrar_id, registrar_name FROM "Registrar";''') AS registrar_table
+ON
+    counts_table.clientId = registrar_table.registrar_id

core/src/test/resources/google/registry/reporting/icann/transaction_transfer_losing_test_cloud_sql.sql

@@ -0,0 +1,72 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Counts the number of mutating transactions each registrar made.
+
+  -- We populate the fields through explicit logging of
+  -- DomainTransactionRecords, which contain all necessary information for
+  -- reporting (such as reporting time, report field, report amount, etc.
+
+SELECT
+    tld,
+    registrar_table.registrar_name AS registrar_name,
+    metricName,
+    metricValue
+FROM (
+    SELECT
+        tld,
+        clientId,
+        CASE WHEN field = 'TRANSFER_SUCCESSFUL' THEN 'TRANSFER_LOSING_SUCCESSFUL'
+        WHEN field = 'TRANSFER_NACKED' THEN 'TRANSFER_LOSING_NACKED'
+        ELSE NULL
+        END AS metricName,
+        SUM(amount) AS metricValue
+    FROM (
+        SELECT
+            CASE
+                -- Explicit transfer acks (approve) and nacks (reject) are done
+                -- by the opposing registrar. Thus, for these specific actions,
+                -- we swap the 'history_other_registrar_id' with the
+                -- 'history_registrar_id' to properly account for this reversal.
+                WHEN (history_type = 'DOMAIN_TRANSFER_APPROVE'
+                OR history_type = 'DOMAIN_TRANSFER_REJECT')
+                THEN history_registrar_id
+                ELSE history_other_registrar_id
+            END AS clientId,
+            tld,
+            report_field AS field,
+            report_amount AS amount,
+            reporting_time AS reportingTime
+            FROM EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+            ''' SELECT history_type, history_other_registrar_id, history_registrar_id, domain_repo_id, history_revision_id FROM "DomainHistory";''') AS dh
+            JOIN
+                EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+                '''SELECT domain_repo_id, history_revision_id, reporting_time, tld, report_field, report_amount FROM "DomainTransactionRecord";''') AS dtr
+            ON
+            dh.domain_repo_id = dtr.domain_repo_id AND dh.history_revision_id = dtr.history_revision_id
+        )
+    -- Only look at this month's data
+    WHERE reportingTime
+    BETWEEN TIMESTAMP('2017-09-01 00:00:00.000')
+    AND TIMESTAMP('2017-09-30 23:59:59.999')
+    GROUP BY
+        tld,
+        clientId,
+        field ) AS counts_table
+JOIN
+     EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql",
+                '''SELECT registrar_id, registrar_name FROM "Registrar";''') AS registrar_table
+ON
+    counts_table.clientId = registrar_table.registrar_id

core/src/test/resources/google/registry/reporting/icann/transactions_report_aggregation_test_cloud_sql.sql

@@ -0,0 +1,97 @@
+#standardSQL
+  -- Copyright 2021 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- Construct the transaction reports' rows from the intermediary data views.
+
+  -- This query pulls from all intermediary tables to create the activity
+  -- report csv, via a table transpose and sum over all activity report fields.
+
+SELECT
+  tlds.tld_name as tld,
+  -- Surround registrar names with quotes to handle names containing a comma.
+  FORMAT("\"%s\"", registrars.registrar_name) as registrar_name,
+  registrars.iana_id as iana_id,
+  SUM(IF(metrics.metricName = 'TOTAL_DOMAINS', metrics.metricValue, 0)) AS total_domains,
+  SUM(IF(metrics.metricName = 'TOTAL_NAMESERVERS', metrics.metricValue, 0)) AS total_nameservers,
+  SUM(IF(metrics.metricName = 'NET_ADDS_1_YR', metrics.metricValue, 0)) AS net_adds_1_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_2_YR', metrics.metricValue, 0)) AS net_adds_2_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_3_YR', metrics.metricValue, 0)) AS net_adds_3_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_4_YR', metrics.metricValue, 0)) AS net_adds_4_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_5_YR', metrics.metricValue, 0)) AS net_adds_5_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_6_YR', metrics.metricValue, 0)) AS net_adds_6_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_7_YR', metrics.metricValue, 0)) AS net_adds_7_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_8_YR', metrics.metricValue, 0)) AS net_adds_8_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_9_YR', metrics.metricValue, 0)) AS net_adds_9_yr,
+  SUM(IF(metrics.metricName = 'NET_ADDS_10_Yr', metrics.metricValue, 0)) AS net_adds_10_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_1_YR', metrics.metricValue, 0)) AS net_renews_1_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_2_YR', metrics.metricValue, 0)) AS net_renews_2_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_3_YR', metrics.metricValue, 0)) AS net_renews_3_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_4_YR', metrics.metricValue, 0)) AS net_renews_4_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_5_YR', metrics.metricValue, 0)) AS net_renews_5_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_6_YR', metrics.metricValue, 0)) AS net_renews_6_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_7_YR', metrics.metricValue, 0)) AS net_renews_7_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_8_YR', metrics.metricValue, 0)) AS net_renews_8_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_9_YR', metrics.metricValue, 0)) AS net_renews_9_yr,
+  SUM(IF(metrics.metricName = 'NET_RENEWS_10_YR', metrics.metricValue, 0)) AS net_renews_10_yr,
+  SUM(IF(metrics.metricName = 'TRANSFER_GAINING_SUCCESSFUL', metrics.metricValue, 0)) AS transfer_gaining_successful,
+  SUM(IF(metrics.metricName = 'TRANSFER_GAINING_NACKED', metrics.metricValue, 0)) AS transfer_gaining_nacked,
+  SUM(IF(metrics.metricName = 'TRANSFER_LOSING_SUCCESSFUL', metrics.metricValue, 0)) AS transfer_losing_successful,
+  SUM(IF(metrics.metricName = 'TRANSFER_LOSING_NACKED', metrics.metricValue, 0)) AS transfer_losing_nacked,
+  -- We don't interact with transfer disputes
+  0 AS transfer_disputed_won,
+  0 AS transfer_disputed_lost,
+  0 AS transfer_disputed_nodecision,
+  SUM(IF(metrics.metricName = 'DELETED_DOMAINS_GRACE', metrics.metricValue, 0)) AS deleted_domains_grace,
+  SUM(IF(metrics.metricName = 'DELETED_DOMAINS_NOGRACE', metrics.metricValue, 0)) AS deleted_domains_nograce,
+  SUM(IF(metrics.metricName = 'RESTORED_DOMAINS', metrics.metricValue, 0)) AS restored_domains,
+  -- We don't require restore reports
+  0 AS restored_noreport,
+  -- We don't enforce AGP limits right now
+  0 AS agp_exemption_requests,
+  0 AS agp_exemptions_granted,
+  0 AS agp_exempted_domains,
+  SUM(IF(metrics.metricName = 'ATTEMPTED_ADDS', metrics.metricValue, 0)) AS attempted_adds
+FROM
+ -- Only produce reports for real TLDs
+    EXTERNAL_QUERY("projects/domain-registry-alpha/locations/us/connections/domain-registry-alpha-sql", '''SELECT "tld_name" FROM "Tld" AS t WHERE t.tld_type='REAL';''') AS tlds
+JOIN
+(SELECT *
+  FROM `domain-registry-alpha.cloud_sql_icann_reporting.registrar_iana_id_201709`)
+  AS registrars
+ON tlds.tld_name = registrars.tld
+-- We LEFT JOIN to produce reports even if the registrar made no transactions
+LEFT OUTER JOIN (
+  -- Gather all intermediary data views
+  SELECT *
+  FROM `domain-registry-alpha.cloud_sql_icann_reporting.total_domains_201709`
+  UNION ALL
+  SELECT *
+  FROM `domain-registry-alpha.cloud_sql_icann_reporting.total_nameservers_201709`
+  UNION ALL
+  SELECT *
+  FROM `domain-registry-alpha.cloud_sql_icann_reporting.transaction_counts_201709`
+  UNION ALL
+  SELECT *
+  FROM `domain-registry-alpha.cloud_sql_icann_reporting.transaction_transfer_losing_201709`
+  UNION ALL
+  SELECT *
+  FROM `domain-registry-alpha.cloud_sql_icann_reporting.attempted_adds_201709` ) AS metrics
+-- Join on tld and registrar name
+ON registrars.tld = metrics.tld
+AND registrars.registrar_name = metrics.registrar_name
+GROUP BY
+tld, registrar_name, iana_id
+ORDER BY
+tld, registrar_name