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 testMappingStructRowFromInitialPartitionToChildPartitionRecord() {
    final Struct struct =
        recordsToStructWithStrings(
            new ChildPartitionsRecord(
                Timestamp.ofTimeSecondsAndNanos(10L, 20),
                "1",
                Arrays.asList(
                    new ChildPartition("childToken1", Sets.newHashSet()),
                    new ChildPartition("childToken2", Sets.newHashSet())),
                null));
    final ChildPartitionsRecord expected =
        new ChildPartitionsRecord(
            Timestamp.ofTimeSecondsAndNanos(10L, 20),
            "1",
            Arrays.asList(
                new ChildPartition(
                    "childToken1", Sets.newHashSet(InitialPartition.PARTITION_TOKEN)),
                new ChildPartition(
                    "childToken2", Sets.newHashSet(InitialPartition.PARTITION_TOKEN))),
            null);
    ChangeStreamResultSet resultSet = mock(ChangeStreamResultSet.class);
    when(resultSet.getCurrentRowAsStruct()).thenReturn(struct);

    final PartitionMetadata initialPartition =
        partition.toBuilder().setPartitionToken(InitialPartition.PARTITION_TOKEN).build();

    assertEquals(
        Collections.singletonList(expected),
        mapper.toChangeStreamRecords(initialPartition, resultSet, resultSetMetadata));
  }