### Question:
MergeStatementBuildingFn extends DoFn<KV<String, KV<Schema, Schema>>, KV<String, BigQueryAction>> { public static String buildQueryGetMaximumTimestampPerPrimaryKey( String tableName, List<String>pkColumns, String projectId, String datasetId) { String pkCommaJoinedString = String.join(", ", pkColumns); String maxTimestampPerPkStatement = String.format( MAXIMUM_TIMESTAMP_PER_PK, pkCommaJoinedString, DataflowCdcRowFormat.TIMESTAMP_MS, projectId, datasetId, tableName, pkCommaJoinedString); return maxTimestampPerPkStatement; } 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 testGetMaximumTimestampPerPrimaryKey() { String getMaximumTsPkQuery = MergeStatementBuildingFn.buildQueryGetMaximumTimestampPerPrimaryKey( TABLE_ID, PRIMARY_KEY_COLUMNS, PROJECT_ID, CHANGELOG_DATASET_ID); assertThat(getMaximumTsPkQuery, equalTo(String.join("", "SELECT pk1, pk2, MAX(timestampMs) as max_ts_ms FROM ", "`myProject.myChangelogDataset.myTable` GROUP BY pk1, pk2"))); }