public List<ChangeStreamRecord> toChangeStreamRecords(
      PartitionMetadata partition,
      ChangeStreamResultSet resultSet,
      ChangeStreamResultSetMetadata resultSetMetadata) {
    if (this.isPostgres()) {
      // In PostgresQL, change stream records are returned as JsonB.
      return Collections.singletonList(
          toChangeStreamRecordJson(partition, resultSet.getPgJsonb(0), resultSetMetadata));
    }
    // In GoogleSQL, change stream records are returned as an array of structs.
    return resultSet.getCurrentRowAsStruct().getStructList(0).stream()
        .flatMap(struct -> toChangeStreamRecord(partition, struct, resultSetMetadata))
        .collect(Collectors.toList());
  }
@Test
  public void testMappingInsertJsonRowNewValuesToDataChangeRecord() {
    final DataChangeRecord dataChangeRecord =
        new DataChangeRecord(
            "partitionToken",
            Timestamp.ofTimeSecondsAndNanos(10L, 20),
            "transactionId",
            false,
            "1",
            "tableName",
            Arrays.asList(
                new ColumnType("column1", new TypeCode("{\"code\":\"INT64\"}"), true, 1L),
                new ColumnType("column2", new TypeCode("{\"code\":\"BYTES\"}"), false, 2L)),
            Collections.singletonList(
                new Mod("{\"column1\":\"value1\"}", null, "{\"column2\":\"newValue2\"}")),
            ModType.INSERT,
            ValueCaptureType.NEW_VALUES,
            10L,
            2L,
            "transactionTag",
            true,
            null);
    final String jsonString = recordToJson(dataChangeRecord, false, false);

    assertNotNull(jsonString);
    ChangeStreamResultSet resultSet = mock(ChangeStreamResultSet.class);
    when(resultSet.getPgJsonb(0)).thenReturn(jsonString);
    assertEquals(
        Collections.singletonList(dataChangeRecord),
        mapperPostgres.toChangeStreamRecords(partition, resultSet, resultSetMetadata));
  }