public static <T, PartitionColumnT> ReadWithPartitions<T, PartitionColumnT> readWithPartitions(
      TypeDescriptor<PartitionColumnT> partitioningColumnType) {
    return new AutoValue_JdbcIO_ReadWithPartitions.Builder<T, PartitionColumnT>()
        .setPartitionColumnType(partitioningColumnType)
        .setNumPartitions(DEFAULT_NUM_PARTITIONS)
        .setFetchSize(DEFAULT_FETCH_SIZE)
        .setUseBeamSchema(false)
        .build();
  }
@Test
  public void testPartitioningStringsWithCustomPartitionsHelper() {
    JdbcReadWithPartitionsHelper<String> helper =
        new JdbcReadWithPartitionsHelper<String>() {
          @Override
          public Iterable<KV<String, String>> calculateRanges(
              String lowerBound, String upperBound, Long partitions) {
            // we expect the elements in the test case follow the format <common prefix>idx
            String prefix = StringUtils.getCommonPrefix(lowerBound, upperBound);
            int minChar = lowerBound.charAt(prefix.length());
            int maxChar = upperBound.charAt(prefix.length());
            int numPartition;
            if (maxChar - minChar < partitions) {
              LOG.warn(
                  "Partition large than possible! Adjust to {} partition instead",
                  maxChar - minChar);
              numPartition = maxChar - minChar;
            } else {
              numPartition = Math.toIntExact(partitions);
            }
            List<KV<String, String>> ranges = new ArrayList<>();
            int stride = (maxChar - minChar) / numPartition + 1;
            int highest = minChar;
            for (int i = minChar; i < maxChar - stride; i += stride) {
              ranges.add(KV.of(prefix + (char) i, prefix + (char) (i + stride)));
              highest = i + stride;
            }
            if (highest <= maxChar) {
              ranges.add(KV.of(prefix + (char) highest, prefix + (char) (highest + stride)));
            }
            return ranges;
          }

          @Override
          public void setParameters(
              KV<String, String> element, PreparedStatement preparedStatement) {
            try {
              preparedStatement.setString(1, element.getKey());
              preparedStatement.setString(2, element.getValue());
            } catch (SQLException e) {
              throw new RuntimeException(e);
            }
          }

          @Override
          public KV<Long, KV<String, String>> mapRow(ResultSet resultSet) throws Exception {
            if (resultSet.getMetaData().getColumnCount() == 3) {
              return KV.of(
                  resultSet.getLong(3), KV.of(resultSet.getString(1), resultSet.getString(2)));
            } else {
              return KV.of(0L, KV.of(resultSet.getString(1), resultSet.getString(2)));
            }
          }
        };

    PCollection<TestRow> rows =
        pipeline.apply(
            JdbcIO.<TestRow, String>readWithPartitions(helper)
                .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION)
                .withRowMapper(new JdbcTestHelper.CreateTestRowOfNameAndId())
                .withTable(READ_TABLE_NAME)
                .withNumPartitions(5)
                .withPartitionColumn("name"));
    PAssert.thatSingleton(rows.apply("Count All", Count.globally())).isEqualTo(1000L);
    pipeline.run();
  }