[Bug]: After JDBCIO read withRowOutput(), the VARCHAR/TEXT -> LOGICAL_TYPE and not compatible with SqlTypeName

[haoyuche]

What happened?

    PCollection<Row> rows = pipeline.apply(JdbcIO.<Row>readWithPartitions()
            .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                    input.driverClassName, input.url)
                    .withUsername(input.username)
                    .withPassword(input.password))
            .withTable(input.tableName)
            .withPartitionColumn(input.primeKeyForPartitionRead)
            .withRowOutput());

Now there is a column named "name" which is VARCHAR, which has been read as beam type LOGICAL_TYPE NOT NULL.

Then I do

        Schema schemaAfterSql = Schema.builder().addNullableField("id", Schema.FieldType.INT64).addNullableField("name", Schema.FieldType.String).build();
        String selectQuery = "SELECT id, name FROM PCOLLECTION" ;
        PCollection<Row> processedRows = rows.apply("EnforcePolicy", SqlTransform.query(selectQuery)).setRowSchema(schemaAfterSql);

It gets "An exception occured while executing the Java class. Unable to parse query SELECT id, policy_name FROM PCOLLECTION: java.lang.IllegalArgumentException: Cannot find a matching Calcite SqlTypeName for Beam type: LOGICAL_TYPE NOT NULL"

Issue Priority

Priority: 0

Issue Component

Component: io-java-jdbc

[Abacn]

Thanks for reporting this. It looks like the cause is CalciteUtil here:

beam/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/utils/CalciteUtils.java

Line 224 in 4ffeae4

default:

does not consider the logical type FieldType.

[Abacn]

.remove-label P0, "awaiting triage"

.add_label P2

[Abacn]

.remove-labels P0, "awaiting triage"

.set_labels P2

[Abacn]

.remove-labels 'awaiting triage'

[github-actions]

Label cannot be managed because it does not exist in the repo. Please check your spelling.

[haoyuche]

Yes, I think somehow we should not feed LOGICAL_TYPE to CalciteUtils but instead we should feed the BaseT so that it can handle the normal sql types. So the fix should be within apache/beam/tree/master/sdks/java/extensions/sql/jdbc when parsing the sql query.