JdbcUtil { static JdbcIO.PreparedStatementSetCaller getPreparedStatementSetCaller( Schema.FieldType fieldType) { switch (fieldType.getTypeName()) { case ARRAY: case ITERABLE: return (element, ps, i, fieldWithIndex) -> ps.setArray( i + 1, ps.getConnection() .createArrayOf( fieldType.getCollectionElementType().getTypeName().name(), element.getArray(fieldWithIndex.getIndex()).toArray())); case LOGICAL_TYPE: { String logicalTypeName = fieldType.getLogicalType().getIdentifier(); JDBCType jdbcType = JDBCType.valueOf(logicalTypeName); switch (jdbcType) { case DATE: return (element, ps, i, fieldWithIndex) -> ps.setDate( i + 1, new Date( getDateOrTimeOnly( element.getDateTime(fieldWithIndex.getIndex()).toDateTime(), true) .getTime() .getTime())); case TIME: return (element, ps, i, fieldWithIndex) -> ps.setTime( i + 1, new Time( getDateOrTimeOnly( element.getDateTime(fieldWithIndex.getIndex()).toDateTime(), false) .getTime() .getTime())); case TIMESTAMP_WITH_TIMEZONE: return (element, ps, i, fieldWithIndex) -> { Calendar calendar = withTimestampAndTimezone( element.getDateTime(fieldWithIndex.getIndex()).toDateTime()); ps.setTimestamp(i + 1, new Timestamp(calendar.getTime().getTime()), calendar); }; default: return getPreparedStatementSetCaller(fieldType.getLogicalType().getBaseType()); } } default: { if (typeNamePsSetCallerMap.containsKey(fieldType.getTypeName())) { return typeNamePsSetCallerMap.get(fieldType.getTypeName()); } else { throw new RuntimeException( fieldType.getTypeName().name() + " in schema is not supported while writing. Please provide statement and preparedStatementSetter"); } } } }    }
@Test public void testGetPreparedStatementSetCaller() throws Exception { Schema wantSchema = Schema.builder() .addField("col1", Schema.FieldType.INT64) .addField("col2", Schema.FieldType.INT64) .addField("col3", Schema.FieldType.INT64) .build(); String generatedStmt = JdbcUtil.generateStatement("test_table", wantSchema.getFields()); String expectedStmt = "INSERT INTO test_table(col1, col2, col3) VALUES(?, ?, ?)"; assertEquals(expectedStmt, generatedStmt); }