JdbcIO { @Experimental(Kind.SCHEMAS) public static ReadRows readRows() { return new AutoValue_JdbcIO_ReadRows.Builder() .setFetchSize(DEFAULT_FETCH_SIZE) .setOutputParallelization(true) .setStatementPreparator(ignored -> {}) .build(); } private  JdbcIO(); static Read<T> read(); @Experimental(Kind.SCHEMAS) static ReadRows readRows(); static ReadAll<ParameterT, OutputT> readAll(); static Write<T> write(); static WriteVoid<T> writeVoid();  }
@Test public void testReadRowsWithDataSourceConfiguration() { PCollection<Row> rows = pipeline.apply( JdbcIO.readRows() .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION) .withQuery(String.format("select name,id from %s where name = ?", READ_TABLE_NAME)) .withStatementPreparator( preparedStatement -> preparedStatement.setString(1, TestRow.getNameForSeed(1)))); Schema expectedSchema = Schema.of( Schema.Field.of("NAME", LogicalTypes.variableLengthString(JDBCType.VARCHAR, 500)) .withNullable(true), Schema.Field.of("ID", Schema.FieldType.INT32).withNullable(true)); assertEquals(expectedSchema, rows.getSchema()); PCollection<Row> output = rows.apply(Select.fieldNames("NAME", "ID")); PAssert.that(output) .containsInAnyOrder( ImmutableList.of(Row.withSchema(expectedSchema).addValues("Testval1", 1).build())); pipeline.run(); }
@Test public void testReadRowsWithoutStatementPreparator() { SerializableFunction<Void, DataSource> dataSourceProvider = ignored -> DATA_SOURCE; String name = TestRow.getNameForSeed(1); PCollection<Row> rows = pipeline.apply( JdbcIO.readRows() .withDataSourceProviderFn(dataSourceProvider) .withQuery( String.format( "select name,id from %s where name = '%s'", READ_TABLE_NAME, name))); Schema expectedSchema = Schema.of( Schema.Field.of("NAME", LogicalTypes.variableLengthString(JDBCType.VARCHAR, 500)) .withNullable(true), Schema.Field.of("ID", Schema.FieldType.INT32).withNullable(true)); assertEquals(expectedSchema, rows.getSchema()); PCollection<Row> output = rows.apply(Select.fieldNames("NAME", "ID")); PAssert.that(output) .containsInAnyOrder( ImmutableList.of(Row.withSchema(expectedSchema).addValues(name, 1).build())); pipeline.run(); }
@Test public void testReadRowsWithDataSourceConfiguration() { PCollection<Row> rows = pipeline.apply( JdbcIO.readRows() .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(dataSource)) .withQuery(String.format("select name,id from %s where name = ?", readTableName)) .withStatementPreparator( preparedStatement -> preparedStatement.setString(1, TestRow.getNameForSeed(1)))); Schema expectedSchema = Schema.of( Schema.Field.of("NAME", LogicalTypes.variableLengthString(JDBCType.VARCHAR, 500)) .withNullable(true), Schema.Field.of("ID", Schema.FieldType.INT32).withNullable(true)); assertEquals(expectedSchema, rows.getSchema()); PCollection<Row> output = rows.apply(Select.fieldNames("NAME", "ID")); PAssert.that(output) .containsInAnyOrder( ImmutableList.of(Row.withSchema(expectedSchema).addValues("Testval1", 1).build())); pipeline.run(); }
@Test public void testReadRowsWithoutStatementPreparator() { SerializableFunction<Void, DataSource> dataSourceProvider = ignored -> dataSource; String name = TestRow.getNameForSeed(1); PCollection<Row> rows = pipeline.apply( JdbcIO.readRows() .withDataSourceProviderFn(dataSourceProvider) .withQuery( String.format( "select name,id from %s where name = '%s'", readTableName, name))); Schema expectedSchema = Schema.of( Schema.Field.of("NAME", LogicalTypes.variableLengthString(JDBCType.VARCHAR, 500)) .withNullable(true), Schema.Field.of("ID", Schema.FieldType.INT32).withNullable(true)); assertEquals(expectedSchema, rows.getSchema()); PCollection<Row> output = rows.apply(Select.fieldNames("NAME", "ID")); PAssert.that(output) .containsInAnyOrder( ImmutableList.of(Row.withSchema(expectedSchema).addValues(name, 1).build())); pipeline.run(); }