[Bug]: Using string in PartitionColumn throws error, tries to convert it to string

[ssaurav-redhat]

What happened?

Referencing JdbcIo's doc in the section for Parallel reading from a JDBC datasource, It mentions to use either of these types of column for paritioning Beam supports partitioned reading of all data from a table. Automatic partitioning is supported for a few data types: Long, [DateTime](https://static.javadoc.io/joda-time/joda-time/2.10.10/org/joda/time/DateTime.html?is-external=true), String.

But when I am passing a string field into it, it tries to convert it into Long. As a result the code fails. Can anyone help me with this?

Beam version -> 2.56.0 JdbcIo version -> 2.56.0

Code PCollection<LogTable> dbReadData = p.apply("ReadFromMySQL", JdbcIO.<LogTable>readWithPartitions() .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create( "com.mysql.cj.jdbc.Driver", "jdbc:mysql://localhost:3307/exampledb") .withUsername("exampleuser") .withPassword("examplepass")) .withTable("logs_table") .withCoder(SerializableCoder.of(LogTable.class)) .withPartitionColumn("uuid") .withRowMapper(new JdbcIO.RowMapper<LogTable>() { @Override public LogTable mapRow(ResultSet resultSet) throws Exception { String uuid = resultSet.getString("uuid"); Timestamp time = resultSet.getTimestamp("time"); return new LogTable(uuid, time); } }));

table structure -> CREATE TABLElogs_table( uuidvarchar(255) NOT NULL, timetimestamp NOT NULL )

Runner used was Flink Runner

Issue Priority

Priority: 3 (minor)

Issue Components

• [ ] Component: Python SDK
• [X] Component: Java SDK
• [ ] Component: Go SDK
• [ ] Component: Typescript SDK
• [ ] Component: IO connector
• [ ] Component: Beam YAML
• [ ] Component: Beam examples
• [ ] Component: Beam playground
• [ ] Component: Beam katas
• [ ] Component: Website
• [ ] Component: Spark Runner
• [X] Component: Flink Runner
• [ ] Component: Samza Runner
• [ ] Component: Twister2 Runner
• [ ] Component: Hazelcast Jet Runner
• [ ] Component: Google Cloud Dataflow Runner

[liferoad]

@Abacn is this expected?

[albmargareugen]

I'm facing this error using Spark Runner. The field id is also a String:

Exception in thread "main" java.lang.NullPointerException: readWithPartitions only supports the following types: [class java.lang.Long, class org.joda.time.DateTime] at org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull(Preconditions.java:1010) at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadWithPartitions.expand(JdbcIO.java:1355) at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadWithPartitions.expand(JdbcIO.java:1189)

public static JdbcIO.ReadWithPartitions<TableDB, String> generateTableWithPartition(JdbcIO.DataSourceConfiguration ds) { return JdbcIO.<TableDB, String>readWithPartitions(TypeDescriptor.of(String.class)) .withDataSourceConfiguration(ds) .withTable("table") .withPartitionColumn("id") .withRowMapper(new RowMapperGenerator.CreateTableSource()); }

I had took a look into here and line 492 defines public static final Map<Class<?>, JdbcReadWithPartitionsHelper<?>> PRESET_HELPERS, which includes the types that the partition can actually handle. The only class types defined there are Long and DateTime, not the String...

However the documentation says explicitly that it can handle the type String in here at line 120!

[Abacn]

Read the code, as early as when this documentation was entered #15848, it never supported String type as partition type.

[albmargareugen]

Thanks Abacn! Sad it's not implemented, but at least now I know why.

[Abacn]

dup #27120