### Question:
MergeStatementBuildingFn extends DoFn<KV<String, KV<Schema, Schema>>, KV<String, BigQueryAction>> { public static String buildQueryMergeReplicaTableWithChangeLogTable( String replicaTableName, String changelogTableName, List<String>pkColumns, List<String> rowColumns, String projectId, String changelogDatasetId, String replicaDatasetId) { String changeLogLatestChangePerKey = buildQueryGetLatestChangePerPrimaryKey( changelogTableName, pkColumns, projectId, changelogDatasetId); String finalMergeQuery = String.format(MERGE_REPLICA_WITH_CHANGELOG_TABLES, projectId, replicaDatasetId, replicaTableName, changeLogLatestChangePerKey, buildJoinConditions( pkColumns, "replica", "changelog" + ".primaryKey"), buildUpdateStatement(rowColumns), buildInsertStatement(rowColumns, rowColumns.stream().map(col -> "fullRecord." + col).collect(Collectors.toList()))); return finalMergeQuery; } 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 testFullMergeStatement() { String fullMergeStatement = MergeStatementBuildingFn.buildQueryMergeReplicaTableWithChangeLogTable( TABLE_ID, CHANGELOG_TABLE_ID, PRIMARY_KEY_COLUMNS, FULL_COLUMN_LIST, PROJECT_ID, CHANGELOG_DATASET_ID, REPLICA_DATASET_ID); assertThat(fullMergeStatement, containsString("ON " + "replica.pk1 = changelog.primaryKey.pk1 " + "AND replica.pk2 = changelog.primaryKey.pk2 ")); assertThat(fullMergeStatement, containsString("MERGE `myProject.myReplicaDataset.myTable` AS replica")); assertThat(fullMergeStatement, containsString("WHEN MATCHED THEN UPDATE SET " + "pk1 = changelog.fullRecord.pk1, " + "pk2 = changelog.fullRecord.pk2, " + "col1 = changelog.fullRecord.col1, " + "col2 = changelog.fullRecord.col2, " + "col3 = changelog.fullRecord.col3 ")); assertThat(fullMergeStatement, containsString( "WHEN NOT MATCHED BY TARGET AND changelog.operation != \"DELETE\" THEN " + "INSERT(pk1, pk2, col1, col2, col3) " + "VALUES (" + "fullRecord.pk1, fullRecord.pk2, fullRecord.col1, fullRecord.col2, fullRecord.col3)" )); }