### Question:
MergeStatementBuildingFn extends DoFn<KV<String, KV<Schema, Schema>>, KV<String, BigQueryAction>> { public static String buildQueryGetLatestChangePerPrimaryKey( String tableName, List<String>pkColumns, String projectId, String datasetId) { List<String> qualifiedPkColumns = pkColumns. stream() .map(s -> String.format("primaryKey.%s", s)) .collect(Collectors.toList()); String maxTimestampPerPkStatement = buildQueryGetMaximumTimestampPerPrimaryKey( tableName, qualifiedPkColumns, projectId, datasetId); String joinStatement = buildJoinConditions( pkColumns, "source_table.primaryKey", "ts_table") + " AND source_table.timestampMs = ts_table.max_ts_ms"; String fullStatement = String.format( LATEST_CHANGE_PER_PK, String.join(", ", pkColumns), DataflowCdcRowFormat.TIMESTAMP_MS, maxTimestampPerPkStatement, projectId, datasetId, tableName, joinStatement); return fullStatement; } MergeStatementBuildingFn(
      String changelogDatasetId,
      String replicaDatasetId,
      String projectId); @ProcessElement void processElement(
      ProcessContext c,
      @StateId("table_created") ValueState<Boolean> tableCreated); static String buildQueryMergeReplicaTableWithChangeLogTable(
      String replicaTableName, String changelogTableName,
      List<String>pkColumns, List<String> rowColumns,
      String projectId, String changelogDatasetId, String replicaDatasetId); static String buildQueryGetLatestChangePerPrimaryKey(
      String tableName, List<String>pkColumns, String projectId, String datasetId); static String buildJoinConditions(
      List<String> pkColumns, final String leftTableName, final String rightTableName); static String buildQueryGetMaximumTimestampPerPrimaryKey(
      String tableName, List<String>pkColumns, String projectId, String datasetId);  }### Answer:
@Test void testGetNewestElementPerPrimaryKey() { String getLatestElementPkQuery = MergeStatementBuildingFn.buildQueryGetLatestChangePerPrimaryKey( TABLE_ID, PRIMARY_KEY_COLUMNS, PROJECT_ID, CHANGELOG_DATASET_ID); assertThat(getLatestElementPkQuery, containsString( "(SELECT primaryKey.pk1, primaryKey.pk2, MAX(timestampMs) as max_ts_ms " + "FROM `myProject.myChangelogDataset.myTable` " + "GROUP BY primaryKey.pk1, primaryKey.pk2) AS ts_table")); assertThat(getLatestElementPkQuery, containsString( "ON source_table.primaryKey.pk1 = ts_table.pk1 " + "AND source_table.primaryKey.pk2 = ts_table.pk2 " + "AND source_table.timestampMs = ts_table.max_ts_ms")); assertThat(getLatestElementPkQuery, containsString( "INNER JOIN `myProject.myChangelogDataset.myTable` AS source_table")); }